このページは Cloud Translation API によって翻訳されました。

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

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

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

Pub/Sub トピックの作成

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

[トピック] に移動
[トピックを作成] をクリックします。
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 関数の作成方法については、こちらのドキュメントをご覧ください。
これ以外の項目はすべてデフォルト値のままにします。
[関数をデプロイ] をクリックします。

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

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

[インスタンス] に移動
このチュートリアルで使用しているインスタンスの 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 で変更ログを表示する

Google Cloud コンソールで、Pub/Sub の [サブスクリプション] ページに移動します。

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

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

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

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

Cloud Run 関数に移動します。
関数 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"}}

オプションの Cloud Run 関数のトリガーを使用して Pub/Sub に変更をストリーミングする コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。