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

オプションの Cloud Run 関数のトリガーを使用して Pub/Sub に変更をストリーミングする

このチュートリアルでは、Bigtable change streams to Pub/Sub テンプレートの使用方法（トピックやテンプレートの構成方法など）について説明します。イベントストリームによってトリガーされる Cloud Run 関数を、任意のプログラミング言語で作成することもできます。

このチュートリアルは、Bigtable、コードの記述、イベントストリーミングサービスに精通している技術ユーザーを対象としています。

目標

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

変更ストリームを有効にして Bigtable テーブルを作成する。
Bigtable 変更ストリームスキーマを使用して Pub/Sub トピックを作成する。
テンプレートを使用して、Dataflow の Pub/Sub パイプラインに Bigtable 変更ストリームをデプロイする。
Pub/Sub で直接、または Cloud Run 関数のログでイベントストリームを表示する。

費用

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

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

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

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

始める前に

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

Enable the Dataflow, Cloud Bigtable API, Cloud Bigtable Admin API, Pub/Sub, Cloud Run functions, and Cloud Storage APIs.

Roles required to enable APIs

To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

Enable the APIs

コンソールで Cloud Shell をアクティブにします。 Google Cloud

Cloud Shell をアクティブにする

コンソールの下部にある Google Cloud Cloud Shell セッションが開始し、コマンドラインプロンプトが表示されます。Cloud Shell はシェル環境です。Google Cloud CLI がすでにインストールされており、現在のプロジェクトの値もすでに設定されています。セッションが初期化されるまで数秒かかることがあります。

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

gcloud components update
gcloud components install cbt

Pub/Sub トピックの作成

コンソールで、Pub/Sub の [Topics] ページに移動します。 Google Cloud

[トピック] に移動
[トピックを作成] をクリックします。
ID を bigtable-change-stream-topic に設定します。
[スキーマを使用する] を選択します。

[Pub/Sub スキーマを選択] プルダウンで、[新しいスキーマを作成] をクリックします。スキーマを定義する新しいタブが開きます。

スキーマ ID を bigtable-change-stream-schema に設定します。
スキーマタイプを Avro に設定します。
スキーマ定義として次の内容を貼り付けます。スキーマの詳細については、テンプレートのドキュメントページをご覧ください。

  {
      "name" : "ChangelogEntryMessage",
      "type" : "record",
      "namespace" : "com.google.cloud.teleport.bigtable",
      "fields" : [
        { "name" : "rowKey", "type" : "bytes"},
        {
          "name" : "modType",
          "type" : {
            "name": "ModType",
            "type": "enum",
            "symbols": ["SET_CELL", "DELETE_FAMILY", "DELETE_CELLS", "UNKNOWN"]}
        },
        { "name": "isGC", "type": "boolean" },
        { "name": "tieBreaker", "type": "int"},
        { "name": "columnFamily", "type": "string"},
        { "name": "commitTimestamp", "type" : "long"},
        { "name" : "sourceInstance", "type" : "string"},
        { "name" : "sourceCluster", "type" : "string"},
        { "name" : "sourceTable", "type" : "string"},
        { "name": "column", "type" : ["null", "bytes"]},
        { "name": "timestamp", "type" : ["null", "long"]},
        { "name": "timestampFrom", "type" : ["null", "long"]},
        { "name": "timestampTo", "type" : ["null", "long"]},
        { "name" : "value", "type" : ["null", "bytes"]}
    ]
  }

[作成] をクリックしてスキーマを作成します。

[スキーマの作成] タブを閉じて、スキーマリストを更新し、新しく定義したスキーマを選択します。
[作成] をクリックしてトピックを作成します。

省略可: Cloud Run 関数を作成する

Pub/Sub ストリームを Cloud Run 関数で処理することもできます。

bigtable-change-stream-topic トピックの [詳細] ページで、[Cloud Function をトリガー] をクリックします。
[関数名] フィールドに「bt-ps-tutorial-function」と入力します。
[ソースコード] セクションで [ランタイム] プルダウンをクリックし、目的のランタイムとプログラミング言語を選択します。受信時に変更ストリームを出力する hello world が生成されます。Cloud Run 関数の作成方法については、こちらのドキュメントをご覧ください。
これ以外の項目はすべてデフォルト値のままにします。
[関数をデプロイ] をクリックします。

変更ストリームを有効にしてテーブルを作成する

コンソールで、Bigtable の [**インスタンス**] ページに移動します。 Google Cloud

[インスタンス] に移動
このチュートリアルで使用しているインスタンスの ID をクリックします。

使用可能なインスタンスがない場合は、近くのリージョンにデフォルトの構成でインスタンスを作成します。
左側のナビゲーションパネルで [テーブル] をクリックします。
[テーブルの作成] をクリックします。
テーブルに change-streams-pubsub-tutorial という名前を付けます。
cf という名前の列ファミリーを追加します。
[変更ストリームを有効にする] を選択します。
[作成] をクリックします。

データパイプラインを初期化して変更ストリームを取得する

Bigtable の [テーブル] ページで、テーブル change-streams-pubsub-tutorial を見つけます。
[変更ストリーム] 列で、[接続] をクリックします。
ダイアログで [Pub/Sub] を選択します。
[Dataflow ジョブを作成] をクリックします。
Dataflow の [ジョブの作成] ページで、出力 Pub/Sub トピック名を bigtable-change-stream-topic に設定します。
Bigtable アプリケーションプロファイル ID を default に設定します。
[ジョブを実行] をクリックします。
ジョブのステータスが「開始中」または「実行中」になったら処理を続行します。ジョブがキューに追加されてから約 5 分かかります。

Bigtable にデータを書き込む

Cloud Shell で数行を Bigtable に書き込み、変更ログが一部のデータを Pub/Sub ストリームに書き込めるようにします。ジョブの作成後にデータを書き込んでいれば変更が表示されます。ジョブステータスが「running」になるのを待つ必要はありません。

cbt -instance=BIGTABLE_INSTANCE_ID -project=YOUR_PROJECT_ID \
    set change-streams-pubsub-tutorial user123 cf:col1=abc
cbt -instance=BIGTABLE_INSTANCE_ID -project=YOUR_PROJECT_ID \
    set change-streams-pubsub-tutorial user546 cf:col1=def
cbt -instance=BIGTABLE_INSTANCE_ID -project=YOUR_PROJECT_ID \
    set change-streams-pubsub-tutorial user789 cf:col1=ghi

Pub/Sub で変更ログを表示する

コンソールで、Pub/Sub の [Subscriptions] ページに移動します。 Google Cloud

[サブスクリプション] に移動
トピック bigtable-change-stream-topic に対して自動的に作成されたサブスクリプションをクリックします。bigtable-change-stream-topic-sub という名前になります。
[メッセージ] タブに移動します。
[PULL] をクリックします。
メッセージのリストを探索し、書き込んだデータを表示します。

省略可: Cloud Run 関数のログで変更を確認する

Cloud Run 関数を作成した場合は、ログで変更を確認できます。

コンソールで、Cloud Run functions に移動します。 Google Cloud

Cloud Run functions に移動
関数 bt-ps-tutorial-function をクリックします。
[ログ] タブに移動します。
ログが表示されるように、[重大度] が「情報」以上に設定されていることを確認します。
ログを調べて、書き込んだデータを表示します。

出力は次のようになります。

Pub/Sub message: {"rowKey":"user789","modType":"SET_CELL","isGC":false,"tieBreaker":0,"columnFamily":"cf","commitTimestamp":1695653833064548,"sourceInstance":"YOUR-INSTANCE","sourceCluster":"YOUR-INSTANCE-c1","sourceTable":"change-streams-pubsub-tutorial","column":{"bytes":"col1"},"timestamp":{"long":1695653832278000},"timestampFrom":null,"timestampTo":null,"value":{"bytes":"ghi"}}

クリーンアップ

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

Bigtable テーブルを削除する

コンソールで、Bigtable の [**インスタンス**] ページに移動します。 Google Cloud

[インスタンス] に移動
このチュートリアルで使用しているインスタンスの ID をクリックします。
左側のナビゲーションパネルで [テーブル] をクリックします。
change-streams-pubsub-tutorial テーブルを見つけます。
[編集] をクリックします。
[変更ストリームを有効にする] をオフにします。
[保存] をクリックします。
テーブルのオーバーフローメニューを開きます。
[削除] をクリックし、テーブル名を入力して確定します。

変更ストリームパイプラインを停止する

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

[ジョブ] に移動
ジョブリストからストリーミングジョブを選択します。
ナビゲーションで、[停止] をクリックします。
[ジョブの停止] ダイアログでパイプラインをキャンセルし、[ジョブの停止] をクリックします。

Pub/Sub トピックとサブスクリプションを削除する

コンソールで、Pub/Sub の [Topics] ページに移動します。 Google Cloud

[トピック] に移動
「bigtable-change-stream-topic」トピックを選択します。
[削除] をクリックして、確定します。
サイドバーの [サブスクリプション] をクリックします。
bigtable-change-stream-topic-sub サブスクリプションを選択します。
[削除] をクリックして、確定します。

Cloud Run 関数を削除する

コンソールで、Cloud Run functions に移動します。 Google Cloud

Cloud Run functions に移動
bt-ps-tutorial-function 関数を選択します。
[削除] をクリックして、確定します。