使用選用的 Cloud Run 函式觸發程序,將變更串流傳入 Pub/Sub

本教學課程說明如何使用「Bigtable 變更串流至 Pub/Sub」範本,包括如何設定主題及設定範本。您可以選擇建立由事件串流觸發的 Cloud Run 函式,並使用您選擇的程式設計語言。

本教學課程適用於熟悉 Bigtable、程式碼編寫和事件串流服務的技術人員。

建立 Pub/Sub 主題

  1. 前往 Google Cloud 控制台的 Pub/Sub「主題」頁面。

    前往「主題」

  2. 按一下「建立主題」

  3. 將 ID 設定為 bigtable-change-stream-topic

  4. 選取「使用結構定義」

  5. 在「選取 Pub/Sub 結構定義」下拉式選單中,按一下「建立新結構定義」。系統會開啟新分頁,供您定義結構定義。

    1. 將結構定義 ID 設為 bigtable-change-stream-schema
    2. 將結構定義類型設為「Avro」Avro
    3. 貼上下列內容做為結構定義。如要進一步瞭解架構,請參閱範本說明文件頁面
      {
          "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"]}
        ]
      }
    
    1. 按一下「建立」,建立結構定義。
  6. 關閉「建立結構定義」分頁,重新整理結構定義清單,然後選取新定義的結構定義。

  7. 按一下「建立」,建立主題。

選用:建立 Cloud Run 函式

您可能想使用 Cloud Run 函式處理 Pub/Sub 串流。

  1. bigtable-change-stream-topic 主題的「詳細資料」頁面中,按一下「觸發 Cloud Function」
  2. 在「Function name」(函式名稱) 欄位中,輸入名稱 bt-ps-tutorial-function
  3. 在「Source Code」(原始碼) 區段中,按一下「Runtime」(執行階段) 下拉式選單,然後選取您偏好的執行階段和程式設計語言。系統會產生 hello world,在變更串流傳入時列印出來。詳情請參閱編寫 Cloud Run 函式的說明文件。
  4. 其他欄位皆使用預設值。
  5. 按一下「Deploy function」(部署函式)

建立啟用變更串流的資料表

  1. 前往 Google Cloud 控制台的 Bigtable「Instances」(執行個體) 頁面。

    前往「Instances」(執行個體) 頁面

  2. 按一下您在本教學課程中使用的執行個體 ID。

    如果沒有可用的執行個體,請在您附近的區域建立執行個體,並使用預設設定。

  3. 按一下左側導覽窗格中的「表格」

  4. 按一下「建立資料表」

  5. 將資料表命名為 change-streams-pubsub-tutorial

  6. 新增名為「cf」的資料欄系列。

  7. 選取「啟用變更串流」

  8. 點選「建立」

初始化資料管道,擷取變更串流

  1. 在 Bigtable「Tables」(資料表) 頁面中,找出所需資料表 change-streams-pubsub-tutorial
  2. 在「變更串流」欄中,按一下「連線」
  3. 在對話方塊中選取「Pub/Sub」
  4. 按一下「建立 Dataflow 工作」
  5. 在 Dataflow 的「Create job」(建立工作) 頁面中,將輸出 Pub/Sub 主題名稱設為:bigtable-change-stream-topic
  6. 將 Bigtable 應用程式設定檔 ID 設為 default
  7. 按一下「Run Job」(執行工作)
  8. 請先等待工作狀態顯示為「Starting」(啟動中) 或「Running」(執行中),再繼續操作。 工作排入佇列後,大約需要 5 分鐘。

將一些資料寫入 Bigtable

  1. 在 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 中查看變更記錄

  1. 前往 Google Cloud 控制台的 Pub/Sub「Subscriptions」(訂閱項目) 頁面。

    前往「訂閱項目」頁面

  2. 按一下系統為主題自動建立的訂閱項目 bigtable-change-stream-topic。名稱應為 bigtable-change-stream-topic-sub

  3. 前往「訊息」分頁。

  4. 按一下「Pull」(提取)

  5. 瀏覽訊息清單,並查看您寫入的資料。

    變更 Pub/Sub 中的記錄檔訊息

選用:查看 Cloud Run functions 記錄中的變更

如果您建立的是 Cloud Run functions 函式,可以在記錄中查看變更。

  1. 前往 Google Cloud 控制台的「Cloud Run functions」(Cloud Run 函式)

    前往 Cloud Run functions

  2. 按一下函式 bt-ps-tutorial-function

  3. 前往「記錄」分頁。

  4. 請確認「嚴重程度」至少設為「資訊」,這樣您才能查看記錄。

  5. 探索記錄檔,並查看您寫入的資料。

輸出看起來類似以下內容:

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"}}