Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

Bigtable 変更ストリームを処理する

このチュートリアルでは、Bigtable テーブルの変更ストリームをソースとするデータベース変更のリアルタイムストリーム用のデータパイプラインを Dataflow にデプロイする方法について説明します。パイプラインの出力は Cloud Storage 上の一連のファイルに書き込まれます。

音楽再生アプリケーションのサンプルデータセットを用意しています。このチュートリアルでは、再生された曲を追跡し、一定期間にわたり上位 5 位をランク付けします。

このチュートリアルは、コードの記述とへのデータパイプラインのデプロイに精通している技術ユーザーを対象としています。 Google Cloud

目標

このチュートリアルでは、次の方法を説明します。

変更ストリームを有効にして Bigtable テーブルを作成する。
変更ストリームを変換して出力するパイプラインを Dataflow にデプロイする。
データパイプラインの結果を表示する。

費用

このドキュメントでは、課金対象である次のコンポーネントを使用します。 Google Cloud

料金計算ツールを使うと、予想使用量に基づいて費用の見積もりを生成できます。

新規の Google Cloud ユーザーの方は、無料トライアルをご利用いただける場合があります。

このドキュメントに記載されているタスクの完了後、作成したリソースを削除すると、それ以上の請求は発生しません。詳細については、クリーンアップをご覧ください。

始める前に

Google Cloud CLI をインストールし、フェデレーション ID を使用して gcloud CLI にログインします。ログイン後、初期化するには、次のコマンドを実行して Google Cloud CLI を初期化します。

gcloud init

プロジェクトを作成または選択します Google Cloud 。

プロジェクトを選択または作成するために必要なロール

プロジェクトを選択する: プロジェクトの選択に特定の IAM ロールは必要ありません。ロールが付与されているプロジェクトを選択できます。
プロジェクトを作成する: プロジェクトを作成するには、プロジェクト作成者ロール（roles/resourcemanager.projectCreator）が必要です。これには resourcemanager.projects.create 権限が含まれています。詳しくは、ロールを付与する方法をご覧ください。

Google Cloud プロジェクトを作成します。
```
gcloud projects create PROJECT_ID
```
PROJECT_ID は、作成する Google Cloud プロジェクトの名前に置き換えます。
作成した Google Cloud プロジェクトを選択します。
```
gcloud config set project PROJECT_ID
```
PROJECT_ID は、 Google Cloud プロジェクト名に置き換えます。

プロジェクトで課金が有効になっていることを確認します Google Cloud 。

Dataflow、Cloud Bigtable API、Cloud Bigtable Admin API、Cloud Storage API を有効にします。

API を有効にするために必要なロール

API を有効にするには、serviceusage.services.enable 権限が必要です。プロジェクトを作成した場合は、オーナーロール（roles/owner）を通じてこの権限が付与されている可能性があります。それ以外の場合は、Service Usage 管理者ロール（roles/serviceusage.serviceUsageAdmin）を通じてこの権限を取得できます。ロールを付与する方法をご覧ください。

gcloud services enable dataflow.googleapis.com bigtable.googleapis.com bigtableadmin.googleapis.com storage.googleapis.com

cbt CLI を更新してインストールします。

gcloud components update
gcloud components install cbt

環境を準備する

コードを取得する

サンプルコードを含むリポジトリのクローンを作成します。このリポジトリをすでにダウンロードしている場合は、最新バージョンを pull して取得します。

git clone https://github.com/GoogleCloudPlatform/java-docs-samples.git
cd java-docs-samples/bigtable/beam/change-streams

バケットの作成

Cloud Storage バケットを作成します。

gcloud storage buckets create gs://BUCKET_NAME

BUCKET_NAME を、バケット名の要件を満たすバケット名に置き換えます。

Bigtable インスタンスを作成する

このチュートリアルでは、既存のインスタンスを使用することも、近くのリージョンにデフォルト構成でインスタンスを作成することもできます。

テーブルを作成する

このサンプルアプリケーションは、ユーザーが再生している曲を追跡し、リッスンイベントを Bigtable に保存します。1 つの列ファミリー（cf）と 1 つの列（曲）があり、行キーにユーザー ID を使用するテーブルを作成し、変更ストリームを有効にします。

テーブルを作成します。

gcloud bigtable instances tables create song-rank \
--column-families=cf --change-stream-retention-period=7d \
--instance=BIGTABLE_INSTANCE_ID --project=PROJECT_ID

次のように置き換えます。

PROJECT_ID: 使用しているプロジェクトの ID
BIGTABLE_INSTANCE_ID: 新しいテーブルを含むインスタンスの ID

パイプラインを開始する

このパイプラインは、次の手順で変更ストリームを変換します。

変更ストリームを読み取る
曲名を取得する
曲のリッスンイベントを N 秒のウィンドウにグループ化する
トップ 5 の曲をカウントする
結果を出力する

パイプラインを実行します

mvn compile exec:java -Dexec.mainClass=SongRank \
"-Dexec.args=--project=PROJECT_ID --bigtableProjectId=PROJECT_ID \
--bigtableInstanceId=BIGTABLE_INSTANCE_ID --bigtableTableId=song-rank \
--outputLocation=gs://BUCKET_NAME/ \
--runner=dataflow --region=BIGTABLE_REGION --experiments=use_runner_v2"

BIGTABLE_REGION は、Bigtable インスタンスが配置されているリージョンの ID（us-east5 など）に置き換えます。

パイプラインを理解する

パイプラインの次のコードスニペットは、実行中のコードを理解するのに役立ちます。

変更ストリームの読み取り

このサンプルコードでは、特定の Bigtable インスタンスとテーブルのパラメータを使ってソースストリームを構成しています。

p.apply(
        "Stream from Bigtable",
        BigtableIO.readChangeStream()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId())
            .withAppProfileId(options.getBigtableAppProfile())

    )

曲名の取得

曲が再生されると、曲名は列ファミリー cf と列修飾子 song に書き込まれます。コードは変更ストリームのミューテーションから値を抽出し、パイプラインの次のステップに出力します。

private static class ExtractSongName extends DoFn<KV<ByteString, ChangeStreamMutation>, String> {

  @DoFn.ProcessElement
  public void processElement(ProcessContext c) {

    for (Entry e : Objects.requireNonNull(Objects.requireNonNull(c.element()).getValue())
        .getEntries()) {
      if (e instanceof SetCell) {
        SetCell setCell = (SetCell) e;
        if ("cf".equals(setCell.getFamilyName())
            && "song".equals(setCell.getQualifier().toStringUtf8())) {
          c.output(setCell.getValue().toStringUtf8());
        }
      }
    }
  }
}

トップ 5 の曲のカウント

組み込みの Beam 関数 Count と Top.of を使用して、現在のウィンドウで上位 5 つの曲を取得できます。

.apply(Count.perElement())
.apply("Top songs", Top.of(5, new SongComparator()).withoutDefaults())

結果の出力

このパイプラインは、結果を標準出力とファイルに書き込みます。ファイルの場合は、書き込みが 10 個の要素または 1 分間のセグメントのグループに分割されます。

.apply("Print", ParDo.of(new PrintFn()))
.apply(
    "Collect at least 10 elements or 1 minute of elements",
    Window.<String>into(new GlobalWindows())
        .triggering(
            Repeatedly.forever(
                AfterFirst.of(
                    AfterPane.elementCountAtLeast(10),
                    AfterProcessingTime
                        .pastFirstElementInPane()
                        .plusDelayOf(Duration.standardMinutes(1)
                        )
                )
            ))
        .discardingFiredPanes())
.apply(
    "Output top songs",
    TextIO.write()
        .to(options.getOutputLocation() + "song-charts/")
        .withSuffix(".txt")
        .withNumShards(1)
        .withWindowedWrites()
);

パイプラインを表示する

コンソール Google Cloud で、[Dataflow] ページに移動します。

Dataflow に移動
song-rank で始まる名前のジョブをクリックします。
画面下部の [表示] をクリックして、ログパネルを開きます。
[ワーカーログ] をクリックして、変更ストリームの出力ログをモニタリングします。

注: ジョブの初期化中にログが表示されるまでに数分かかることがあります。

ストリームの書き込み

cbt CLI を使用して、さまざまなユーザーの曲の再生回数を song-rank テーブルに書き込みます。これは、時間の経過とともにストリーミングされる曲の再生をシミュレートするように数分間にわたって書き込むように設計されています。

cbt -instance=BIGTABLE_INSTANCE_ID -project=PROJECT_ID import \
song-rank song-rank-data.csv  column-family=cf batch-size=1

出力を表示する

最も人気のある曲を確認するために、Cloud Storage の出力を読み取ります。

gcloud storage cat gs://BUCKET_NAME/song-charts/GlobalWindow-pane-0-00000-of-00001.txt

出力例:

2023-07-06T19:53:38.232Z [KV{The Wheels on the Bus, 199}, KV{Twinkle, Twinkle, Little Star, 199}, KV{Ode to Joy , 192}, KV{Row, Row, Row Your Boat, 186}, KV{Take Me Out to the Ball Game, 182}]
2023-07-06T19:53:49.536Z [KV{Old MacDonald Had a Farm, 20}, KV{Take Me Out to the Ball Game, 18}, KV{Für Elise, 17}, KV{Ode to Joy , 15}, KV{Mary Had a Little Lamb, 12}]
2023-07-06T19:53:50.425Z [KV{Twinkle, Twinkle, Little Star, 20}, KV{The Wheels on the Bus, 17}, KV{Row, Row, Row Your Boat, 13}, KV{Happy Birthday to You, 12}, KV{Over the Rainbow, 9}]

クリーンアップ

このチュートリアルで使用したリソースについて、Google Cloud アカウントに課金されないようにするには、リソースを含むプロジェクトを削除するか、プロジェクトを維持して個々のリソースを削除します。

プロジェクトを削除する

注意 : プロジェクトを削除すると、次のような影響があります。

プロジェクト内のすべてのものが削除されます。既存のプロジェクトを使用してこのドキュメントのタスクを行った場合、そのプロジェクトを削除すると、プロジェクト内で行った他の作業もすべて削除されます。
カスタムプロジェクト ID が失われます。このプロジェクトを作成したときに、将来使用するカスタムプロジェクト ID を作成した可能性があります。そのプロジェクト ID を使用した URL（例: appspot.com）を保持するには、プロジェクト全体ではなく、プロジェクト内の選択したリソースだけを削除します。

複数のアーキテクチャ、チュートリアル、クイックスタートを実施する予定がある場合は、プロジェクトを再利用するとプロジェクトの割り当て上限を超えないようにすることができます。

プロジェクト Google Cloud を削除します。

gcloud projects delete PROJECT_ID

リソースを個別に削除する

バケットとファイルを削除します。

gcloud storage rm --recursive gs://BUCKET_NAME/

テーブルで変更ストリームを無効にします。

gcloud bigtable instances tables update song-rank --instance=BIGTABLE_INSTANCE_ID \
--clear-change-stream-retention-period

テーブル song-rank を削除します。

cbt -instance=BIGTABLE_INSTANCE_ID -project=PROJECT_ID deletetable song-rank

変更ストリームパイプラインを停止します。
1. ジョブを一覧表示し、ジョブ ID を取得します。
```
gcloud dataflow jobs list --region=BIGTABLE_REGION
```
2. ジョブをキャンセルします。
```
gcloud dataflow jobs cancel JOB_ID --region=BIGTABLE_REGION
```
  JOB_ID は、前のコマンドの後に表示されたジョブ ID に置き換えます。