使用 Java 讀取變更串流

適用於 Java 的 Cloud Bigtable 用戶端程式庫提供低階方法，可處理資料變更記錄。不過，在大多數情況下，我們建議您使用 Dataflow 串流變更，而非使用本頁所述方法，因為 Dataflow 會為您處理分割區分割和合併作業。

事前準備

使用 Java 讀取變更串流前，請務必先熟悉「變更串流總覽」。然後完成下列必要條件。

設定驗證方法

如要在本機開發環境中使用本頁的 Java 範例，請安裝並初始化 gcloud CLI，然後使用使用者憑證設定應用程式預設憑證。

  1. 安裝 Google Cloud CLI。

  2. 如果您使用外部識別資訊提供者 (IdP)，請先 使用聯合身分登入 gcloud CLI。

  3. 如果您使用本機殼層，請為使用者帳戶建立本機驗證憑證： 

    gcloud auth application-default login

    如果您使用 Cloud Shell，則不需要執行這項操作。

    如果系統傳回驗證錯誤，且您使用外部識別資訊提供者 (IdP)，請確認您已 使用聯合身分登入 gcloud CLI。

詳情請參閱「 設定本機開發環境的驗證機制」。

如要瞭解如何為正式環境設定驗證機制，請參閱「 為在 Google Cloud上執行的程式碼設定應用程式預設憑證 」。

啟用變更串流

您必須先啟用資料表的變更串流，才能讀取資料表。您也可以建立新資料表，並啟用變更串流。

必要的角色

如要取得讀取 Bigtable 變更串流所需的權限，請要求管理員授予您下列 IAM 角色。

  • Bigtable 系統管理員 (roles/bigtable.admin) 在包含您要串流變更的資料表所在的 Bigtable 執行個體上

將 Java 用戶端程式庫新增為依附元件

在 Maven pom.xml 檔案中新增類似下列的程式碼。將 VERSION 替換為您使用的用戶端程式庫版本。版本必須為 2.21.0 以上。

<dependencies>
  <dependency>
    <groupId>com.google.cloud</groupId>
    <artifactId>google-cloud-bigtable</artifactId>
    <version>VERSION</version>
  </dependency>
</dependencies>

判斷資料表的分區

如要開始提出 ReadChangeStream 要求，您必須瞭解資料表的分區。可以使用 GenerateInitialChangeStreamPartitions 方法判斷這項資訊。以下範例說明如何使用這個方法取得 ByteStringRanges 資料流，代表資料表中的每個分割區。每個 ByteStringRange 都包含分區的開始和結束鍵。

ServerStream<ByteStringRange> partitionStream =
    client.generateInitialChangeStreamPartitions("MyTable");

處理每個分區的變更

然後，您可以使用 ReadChangeStream 方法處理每個分割區的變更。以下範例說明如何開啟分割區的串流，並從目前時間開始。

ReadChangeStreamQuery query =
    ReadChangeStreamQuery.create("MyTable")
        .streamPartition(partition)
        .startTime(Instant.now());
ServerStream<ChangeStreamRecord> changeStream = client.readChangeStream(query);

ReadChangeStreamQuery 接受下列引數：

  • 串流分區 (必填) - 要從中串流變更的分區
  • 下列其中一項：
    • 開始時間 - 從中開始處理變更的提交時間戳記
    • 接續符記 - 代表可從中繼續串流的位置的符記
  • 結束時間 (選用) - 達到這個時間戳記時，系統會停止處理變更。如未提供值，串流會繼續讀取。
  • 心跳時間長度 (選用) - 沒有新變化時的心跳訊息頻率 (預設為五秒)

變更串流錄製格式

傳回的變更串流記錄是下列三種回應類型之一：

  • ChangeStreamMutation：代表「資料變更記錄」的訊息。

  • CloseStream：表示用戶端應停止從串流讀取資料的訊息。

    • 狀態 - 顯示關閉串流的原因。下列任一項：
      • OK - 指定分區已達結束時間
      • OUT_OF_RANGE - 指定的分區已不存在，表示該分區已分割或合併。每個新分割區都必須建立新的 ReadChangeStream 要求。
    • NewPartitions：提供 OUT_OF_RANGE 回應的更新分割資訊。
    • ChangeStreamContinuationTokens - 用於從相同位置繼續發出新 ReadChangeStream 要求的符記清單。每個NewPartition只能有一個。

  • Heartbeat - 含有可用於檢查串流狀態的資訊，且會定期傳送。

    • EstimatedLowWatermark - 估算指定分割區的低水位線
    • ContinuationToken：從目前位置繼續串流指定分割區的權杖。

資料變更記錄內容

如要瞭解變更串流記錄，請參閱「資料變更記錄的內容」。

處理分區異動

表格的分區變更時，ReadChangeStream 請求會傳回 CloseStream 訊息，其中包含從新分區繼續串流所需的資訊。

如果是分割，這會包含多個新分割區，以及每個分割區對應的 ContinuationToken。如要從相同位置繼續串流播放新分割區，請為每個新分割區發出新的 ReadChangeStream 要求，並附上對應的權杖。

舉例來說，假設您正在串流傳輸分割區 [A,C)，而該分割區又分成兩個分割區 [A,B) 和 [B,C)，則預期會發生下列事件序列：

ReadChangeStream(streamPartition = ByteStringRange(A, C)) receives:
CloseStream(
    Status = OUT_OF_RANGE,
    NewPartitions = List(ByteStringRange(A, B), ByteStringRange(B, C))
    ChangeStreamContinuationTokens = List(foo, bar)
)

如要從相同位置繼續串流每個分割區，請傳送下列 ReadChangeStreamQuery 要求：

ReadChangeStreamQuery queryAB =
    ReadChangeStreamQuery.create("myTable")
        .streamPartition(ByteStringRange(A, B))
        .continuationTokens(List.of(foo));

ReadChangeStreamQuery queryBC =
    ReadChangeStreamQuery.create("myTable")
        .streamPartition(ByteStringRange(B, C))
        .continuationTokens(List.of(bar));

如要合併，並從相同分割區繼續作業，您需要傳送新的 ReadChangeStream 要求，其中包含合併分割區的每個權杖。

舉例來說，如果您串流兩個分割區 ([A,B) 和 [B,C))，並將其合併至分割區 [A,C)，則預期會發生下列事件序列：

ReadChangeStream(streamPartition = ByteStringRange(A, B)) receives:
CloseStream(
    Status = OUT_OF_RANGE,
    NewPartitions = List(ByteStringRange(A, C)),
    ChangeStreamContinuationTokens = List(foo)
)

ReadChangeStream(streamPartition = ByteStringRange(B, C)) receives:
CloseStream(
    Status = OUT_OF_RANGE,
    NewPartitions = List(ByteStringRange(A, C)),
    ChangeStreamContinuationTokens = List(bar)
)

如要從相同位置繼續串流傳輸分割區 [A, C)，請傳送類似下列的 ReadChangeStreamQuery：

ReadChangeStreamQuery query =
    ReadChangeStreamQuery.create("myTable")
        .streamPartition(ByteStringRange(A, C))
        .continuationTokens(List.of(foo, bar));

後續步驟