このチュートリアルでは、Bigtable テーブルの変更ストリームをソースとするデータベース変更のリアルタイム ストリーム用のデータ パイプラインを Dataflow にデプロイする方法について説明します。パイプラインの出力は Cloud Storage 上の一連のファイルに書き込まれます。
音楽再生アプリケーションのサンプル データセットを用意しています。このチュートリアルでは、再生された曲を追跡し、一定期間にわたり上位 5 位をランク付けします。
このチュートリアルは、コードの記述と へのデータ パイプラインのデプロイに精通している技術ユーザーを対象としています。 Google Cloud
目標
このチュートリアルでは、次の方法を説明します。
- 変更ストリームを有効にして Bigtable テーブルを作成する。
- 変更ストリームを変換して出力するパイプラインを Dataflow にデプロイする。
- データ パイプラインの結果を表示する。
費用
このドキュメントでは、課金対象である次のコンポーネントを使用します。 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 プロジェクト名に置き換えます。 - プロジェクトを選択する: プロジェクトの選択に特定の 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 プロジェクト名に置き換えます。 -
cbtCLI を更新してインストールします。gcloud components update gcloud components install cbt
Google Cloud CLI をインストールします。 インストール後、 初期化するには、次のコマンドを実行して Google Cloud CLI を初期化します。
gcloud init外部 ID プロバイダ（IdP）を使用している場合は、まず フェデレーション ID を使用して gcloud CLI にログインする必要があります。
プロジェクトを作成または選択します 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.combigtable.googleapis.com bigtableadmin.googleapis.com storage.googleapis.com
Google Cloud CLI をインストールします。 インストール後、 初期化するには、次のコマンドを実行して Google Cloud CLI を初期化します。
gcloud init外部 ID プロバイダ（IdP）を使用している場合は、まず フェデレーション ID を使用して gcloud CLI にログインする必要があります。
プロジェクトを作成または選択します 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.combigtable.googleapis.com bigtableadmin.googleapis.com storage.googleapis.com
環境を準備する
コードを取得する
サンプルコードを含むリポジトリのクローンを作成します。このリポジトリをすでにダウンロードしている場合は、最新バージョンを pull して取得します。
git clone https://github.com/GoogleCloudPlatform/java-docs-samples.git
cd java-docs-samples/bigtable/beam/change-streams
バケットの作成
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