Vorlage "Bigtable-Änderungsstreams für Pub/Sub"

Die Vorlage "Bigtable-Änderungsstreams für Pub/Sub" ist eine Streamingpipeline, die Bigtable-Datenänderungsdatensätze streamt und sie mithilfe von Dataflow in einem Pub/Sub-Thema veröffentlicht.

Mit einem Bigtable-Änderungsstream können Sie Datenmutationen auf Tabellenbasis abonnieren. Wenn Sie Änderungsstreams für Tabellen abonnieren, gelten die folgenden Einschränkungen:

Es werden nur geänderte Zellen und Deskriptoren von Löschvorgängen zurückgegeben.
Es wird nur der neue Wert einer geänderten Zelle zurückgegeben.

Wenn Datensätze zur Datenänderung in einem Pub/Sub-Thema veröffentlicht werden, könnten Nachrichten in der falschen Reihenfolge eingefügt werden im Vergleich zur ursprünglichen Reihenfolge des Bigtable-Commits.

Bigtable-Datenänderungseinträge, die nicht in Pub/Sub-Themen veröffentlicht werden können, werden vorübergehend in einem Verzeichnis für Dead-Letter-Warteschlangen (unverarbeitete Nachrichtenwarteschlange) in Cloud Storage abgelegt. Nach der maximalen Anzahl fehlgeschlagener Wiederholungsversuche werden diese Einträge für eine manuelle Überprüfung oder weitere Verarbeitung durch den Nutzer auf unbestimmte Zeit im selben Warteschlangenverzeichnis für unzustellbare Nachrichten abgelegt.

Für die Pipeline muss das Pub/Sub-Zielthema vorhanden sein. Das Zielthema kann so konfiguriert werden, dass Nachrichten mithilfe eines Schemas validiert werden. Wenn ein Pub/Sub-Thema ein Schema angibt, wird die Pipeline nur gestartet, wenn das Schema gültig ist. Verwenden Sie je nach Schematyp eine der folgenden Schemadefinitionen für das Zielthema:

Log-Puffer

syntax = "proto2";

package com.google.cloud.teleport.bigtable;

option java_outer_classname = "ChangeLogEntryProto";

message ChangelogEntryProto{
  required bytes rowKey = 1;
  enum ModType {
    SET_CELL = 0;
    DELETE_FAMILY = 1;
    DELETE_CELLS = 2;
    UNKNOWN = 3;
  }
  required ModType modType = 2;
  required bool isGC = 3;
  required int32 tieBreaker = 4;
  required int64 commitTimestamp = 5;
  required string columnFamily = 6;
  optional bytes column = 7;
  optional int64 timestamp = 8;
  optional int64 timestampFrom = 9;
  optional int64 timestampTo = 10;
  optional bytes value = 11;
  required string sourceInstance = 12;
  required string sourceCluster = 13;
  required string sourceTable = 14;
}

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

JSON

Verwenden Sie das folgende Protobuf-Schema mit JSON-Nachrichtencodierung:

syntax = "proto2";

package com.google.cloud.teleport.bigtable;

option java_outer_classname = "ChangelogEntryMessageText";

message ChangelogEntryText{
  required string rowKey = 1;
  enum ModType {
    SET_CELL = 0;
    DELETE_FAMILY = 1;
    DELETE_CELLS = 2;
    UNKNOWN = 3;
  }
  required ModType modType = 2;
  required bool isGC = 3;
  required int32 tieBreaker = 4;
  required int64 commitTimestamp = 5;
  required string columnFamily = 6;
  optional string column = 7;
  optional int64 timestamp = 8;
  optional int64 timestampFrom = 9;
  optional int64 timestampTo = 10;
  optional string value = 11;
  required string sourceInstance = 12;
  required string sourceCluster = 13;
  required string sourceTable = 14;
}

Jede neue Pub/Sub-Nachricht enthält einen Datenänderungseintrag, der vom Änderungsstream aus der entsprechenden Zeile in Ihrer Bigtable-Tabelle zurückgegeben wird.

Beschreibung der Pub/Sub-Ausgabenachricht

Feldname	Beschreibung
`rowKey`	Der Zeilenschlüssel der geänderten Zeile. Sie erhält ein Formular mit einem Byte-Array. Wenn die JSON-Nachrichtencodierung konfiguriert ist, werden Zeilenschlüssel als Strings zurückgegeben. Wenn `useBase64Rowkeys` angegeben ist, werden Zeilenschlüssel Base64-codiert. Andernfalls wird ein durch `bigtableChangeStreamCharset` angegebenes Zeichensatz verwendet, um Zeilenschlüsselbyte in einen String zu decodieren.
`modType`	Der Typ der Zeilenmutation. Verwenden Sie einen der folgenden Werte: `SET_CELL`, `DELETE_CELLS` oder `DELETE_FAMILY`.
`columnFamily`	Die von der Zeilenmutation betroffene Spaltenfamilie
`column`	Der von der Zeilenmutation betroffene Spaltenqualifizierer. Für den Mutationstyp `DELETE_FAMILY` ist das Spaltenfeld nicht festgelegt. Sie erhält ein Formular mit einem Byte-Array. Wenn die JSON-Nachrichtencodierung konfiguriert ist, werden Spalten als Strings zurückgegeben. Wenn `useBase64ColumnQualifier` angegeben ist, ist das Spaltenfeld Base64-codiert. Andernfalls wird ein durch `bigtableChangeStreamCharset` angegebenes Zeichensatz verwendet, um Zeilenschlüsselbyte in einen String zu decodieren.
`commitTimestamp`	Der Zeitpunkt, an dem Bigtable die Mutation anwendet. Die Zeit wird in Mikrosekunden seit der Unix-Epoche gemessen (1. Januar 1970 um UTC).
`timestamp`	Der Zeitstempelwert der Zelle, die von der Mutation betroffen ist. Für die Mutationstypen `DELETE_CELLS` und `DELETE_FAMILY` ist kein Zeitstempel festgelegt. Die Zeit wird in Mikrosekunden seit der Unix-Epoche gemessen (1. Januar 1970 um UTC).
`timestampFrom`	Beschreibt einen inklusiven Start des Zeitstempelintervalls für alle Zellen, die von der Mutation `DELETE_CELLS` gelöscht wurden. Bei anderen Mutationstypen ist `timestampFrom` nicht festgelegt. Die Zeit wird in Mikrosekunden seit der Unix-Epoche gemessen (1. Januar 1970 um UTC).
`timestampTo`	Beschreibt ein exklusives Ende des Zeitstempelintervalls für alle Zellen, die mit der Mutation `DELETE_CELLS` gelöscht werden. Bei anderen Mutationstypen ist `timestampTo` nicht festgelegt.
`isGC`	Ein boolescher Wert, mit dem angegeben wird, ob die Mutation von einem Speicherbereinigungsmechanismus von Bigtable generiert wird.
`tieBreaker`	Wenn zwei Mutationen von verschiedenen Bigtable-Clustern gleichzeitig registriert werden, wird die Mutation mit dem höchsten `tiebreaker`-Wert auf die Quelltabelle angewendet. Mutationen mit niedrigeren `tiebreaker`-Werten werden verworfen.
`value`	Der neue Wert, der von der Mutation festgelegt wird. Sofern nicht die Pipelineoption `stripValues` festgelegt ist, wird der Wert für `SET_CELL`-Mutationen festgelegt. Bei anderen Mutationstypen wird der Wert nicht festgelegt. Sie erhält ein Formular mit einem Byte-Array. Wenn die JSON-Nachrichtencodierung konfiguriert ist, werden Werte als Strings zurückgegeben. Wenn `useBase64Values` angegeben ist, ist der Wert Base64-codiert. Andernfalls wird ein durch `bigtableChangeStreamCharset` angegebenes Zeichensatz verwendet, um Wertbyte in einen String zu decodieren.
`sourceInstance`	Der Name der Bigtable-Instanz, die die Mutation registriert hat. Dies kann vorkommen, wenn mehrere Pipelines Änderungen von verschiedenen Instanzen zum selben Pub/Sub-Thema streamen.
`sourceCluster`	Der Name des Bigtable-Clusters, der die Mutation registriert hat. Kann verwendet werden, wenn mehrere Pipelines Änderungen von verschiedenen Instanzen zum selben Pub/Sub-Thema streamen.
`sourceTable`	Der Name der Bigtable-Tabelle, die die Mutation erhalten hat. Kann verwendet werden, wenn mehrere Pipelines aus verschiedenen Tabellen in dasselbe Pub/Sub-Thema streamen.

Pipelineanforderungen

Die angegebene Bigtable-Quellinstanz.
Die angegebene Bigtable-Quelltabelle In der Tabelle müssen Änderungsstreams aktiviert sein.
Das angegebene Bigtable-Anwendungsprofil.
Das angegebene Pub/Sub-Thema muss vorhanden sein.

Vorlagenparameter

Parameter	Beschreibung
`bigtableReadInstanceId`	Die Bigtable-Quellinstanz-ID.
`bigtableReadTableId`	Die Bigtable-Quelltabellen-ID.
`bigtableChangeStreamAppProfile`	Die Bigtable-Anwendungsprofil-ID. Das Anwendungsprofil muss Single-Cluster-Routing verwenden und Transaktionen für einzelne Zeilen zulassen.
`pubSubTopic`	Der Name des Pub/Sub-Zielthemas.
`messageFormat`	Optional: Wenn das Schema des Ziels konfiguriert ist, wird das Nachrichtenformat durch das konfigurierte Schema und die Codierung bestimmt. Das Format der Nachrichten, die im Pub/Sub-Thema veröffentlicht werden sollen. Unterstützte Werte: `AVRO`, `PROTOCOL_BUFFERS` und `JSON`. Die Standardeinstellung ist `JSON`. Wenn das `JSON`-Format verwendet wird, sind "rowKey", "column" und "value" der Nachrichten Strings, deren Inhalt durch die Pipelineoptionen `useBase64Rowkeys`, `useBase64ColumnQualifiers`, `useBase64Values` und `bigtableChangeStreamCharset` bestimmt wird.
`messageEncoding`	Optional: Wenn das Schema des Ziels konfiguriert ist, wird die Nachrichtencodierung durch die Themeneinstellungen bestimmt. Die Codierung der Nachrichten, die im Pub/Sub-Thema veröffentlicht werden sollen. Unterstützte Werte: `BINARY`, `JSON` und `JSON`. Die Standardeinstellung ist `JSON`.
`stripValues`	Optional: Wenn dieser Wert auf `true` gesetzt ist, werden die `SET_CELL`-Mutationen ohne neue Werte zurückgegeben. Die Standardeinstellung ist "false". Dieser Parameter ist nützlich, wenn Sie keinen neuen Wert haben müssen, der als Cache-Entwertung bezeichnet wird, oder wenn die Werte extrem groß sind und die Größenbeschränkungen für Pub/Sub-Nachrichten überschreiten.
`bigtableReadProjectId`	Optional: Die Bigtable-Projekt-ID. Der Standardwert ist das Projekt für den Dataflow-Job.
`pubSubProjectId`	Optional: Die Bigtable-Projekt-ID. Der Standardwert ist das Projekt für den Dataflow-Job.
`bigtableChangeStreamMetadataInstanceId`	Optional: Die Bigtable-Änderung streamt die Metadateninstanz-ID.
`bigtableChangeStreamMetadataTableTableId`	Optional: Die Bigtable-Änderung streamt die Metadatentabellen-ID.
`bigtableChangeStreamCharset`	Optional: Die Bigtable-Änderung streamt den Zeichensatznamen beim Lesen von Zeilenschlüsseln, Werten und Spaltenqualifizierern. Diese Option wird verwendet, wenn die Nachrichtencodierung JSON ist.
`bigtableChangeStreamStartTimestamp`	(Optional) Der Start-Zeitstempel (einschließlich), der zum Lesen von Änderungsstreams verwendet wird. Beispiel: `2022-05-05T07:59:59Z`. Die Standardeinstellung ist der Zeitstempel der Pipeline-Startzeit.
`bigtableChangeStreamIgnoreColumnFamilies`	Optional: Eine durch Kommas getrennte Liste von Änderungen der Spaltenfamilien, die ignoriert werden sollen.
`bigtableChangeStreamIgnoreColumns`	Optional: Eine durch Kommas getrennte Liste mit Spaltennamen, die ignoriert werden sollen.
`bigtableChangeStreamName`	Optional: Eindeutiger Name der Clientpipeline. Ermöglicht die Fortsetzung der Verarbeitung ab dem Punkt, an dem eine zuvor ausgeführte Pipeline beendet wurde. Standardmäßig wird der automatisch generierte Name verwendet. Den Wert finden Sie in den Dataflow-Joblogs.
`bigtableChangeStreamResume`	Optional: Wenn dieser Wert auf `true` gesetzt ist, wird eine neue Pipeline ab dem Punkt, an dem eine zuvor ausgeführte Pipeline mit demselben `bigtableChangeStreamName`-Wert beendet wird, mit der Verarbeitung fortgesetzt. Wenn die Pipeline mit dem angegebenen `bigtableChangeStreamName`-Wert noch nie ausgeführt wurde, wird eine neue Pipeline nicht gestartet. Wenn `false` festgelegt ist, wird eine neue Pipeline gestartet. Wenn eine Pipeline mit demselben `bigtableChangeStreamName`-Wert bereits für die angegebene Quelle ausgeführt wurde, wird eine neue Pipeline nicht gestartet. Die Standardeinstellung ist `false`.
`useBase64Rowkeys`	Optional: Wird mit JSON-Nachrichtencodierung verwendet. Wenn `true` festgelegt ist, ist das Feld `rowKey` ein Base64-codierter String. Andernfalls wird `rowKey` mithilfe von `bigtableChangeStreamCharset` erzeugt, um Byte in einen String zu decodieren. Die Standardeinstellung ist `false`.
`useBase64ColumnQualifiers`	Optional: Wird mit JSON-Nachrichtencodierung verwendet. Wenn `true` festgelegt ist, ist das Feld `column` ein Base64-codierter String. Andernfalls wird die Spalte mithilfe von `bigtableChangeStreamCharset` erzeugt, um Byte in einen String zu decodieren. Die Standardeinstellung ist `false`.
`useBase64Values`	Optional: Wird mit JSON-Nachrichtencodierung verwendet. Wenn `true` festgelegt ist, ist das Feld `value` ein Base64-codierter String. Andernfalls wird der Wert mithilfe von `bigtableChangeStreamCharset` generiert, um Byte in einen String zu decodieren. Die Standardeinstellung ist `false`.
`dlqMaxRetries`	Optional: Die maximalen Wiederholungsversuche für unzustellbare Nachrichten. Die Standardeinstellung ist `5`.
`dlqRetryMinutes`	Optional: Die Anzahl der Minuten zwischen DLQ-Wiederholungen (Dead Letter Queue). Die Standardeinstellung ist `10`.
`dlqDirectory`	Optional: Das Verzeichnis für die Warteschlange für unzustellbare Nachrichten. Einträge, die nicht verarbeitet werden können, werden in diesem Verzeichnis gespeichert. Der Standardwert ist ein Verzeichnis unter dem temporären Speicherort des Dataflow-Jobs. In den meisten Fällen können Sie den Standardpfad verwenden.

Führen Sie die Vorlage aus.

Console

Rufen Sie die Dataflow-Seite Job aus Vorlage erstellen auf.

Zur Seite "Job aus Vorlage erstellen“

Geben Sie im Feld Jobname einen eindeutigen Jobnamen ein.
Optional: Wählen Sie für Regionaler Endpunkt einen Wert aus dem Drop-down-Menü aus. Die Standardregion ist us-central1.
Eine Liste der Regionen, in denen Sie einen Dataflow-Job ausführen können, finden Sie unter Dataflow-Standorte.
Wählen Sie im Drop-down-Menü Dataflow-Vorlage die Option the Bigtable change streams to Pub/Sub template aus.
Geben Sie Ihre Parameterwerte in die Parameterfelder ein.
Klicken Sie auf Job ausführen.

gcloud

Führen Sie die Vorlage in der Shell oder im Terminal aus:

gcloud dataflow flex-template run JOB_NAME \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_PubSub \
    --parameters \
bigtableReadInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableReadTableId=BIGTABLE_TABLE_ID,\
bigtableChangeStreamAppProfile=BIGTABLE_APPLICATION_PROFILE_ID,\
pubSubTopic=PUBSUB_TOPIC

Dabei gilt:

PROJECT_ID: die ID des Google Cloud-Projekts, in dem Sie den Dataflow-Job ausführen möchten
JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
VERSION: die Version der Vorlage, die Sie verwenden möchten
Sie können die folgenden Werte verwenden:
- latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
- Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
Achtung: Die neueste Version der Vorlagen wird möglicherweise mit funktionsgefährdenden Änderungen aktualisiert. In Ihren Produktionsumgebungen sollten Vorlagen verwendet werden, die sich im aktuellsten datierten übergeordneten Ordner befinden, um zu verhindern, dass diese funktionsgefährdenden Änderungen Ihre Produktionsworkflows beeinträchtigen.
REGION_NAME: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
BIGTABLE_INSTANCE_ID: Ihre Bigtable-Instanz-ID.
BIGTABLE_TABLE_ID: Ihre Bigtable-Tabellen-ID.
BIGTABLE_APPLICATION_PROFILE_ID: Ihre Bigtable-Anwendungsprofil-ID
PUBSUB_TOPIC: der Name des Pub/Sub-Zielthemas

API

Senden Sie eine HTTP-POST-Anfrage, um die Vorlage mithilfe der REST API auszuführen. Weitere Informationen zur API und ihren Autorisierungsbereichen finden Sie unter projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
  "launch_parameter": {
    "jobName": "JOB_NAME",
    "containerSpecGcsPath": "gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_PubSub",
    "parameters": {
        "bigtableReadInstanceId": "BIGTABLE_INSTANCE_ID",
        "bigtableReadTableId": "BIGTABLE_TABLE_ID",
        "bigtableChangeStreamAppProfile": "BIGTABLE_APPLICATION_PROFILE_ID",
        "pubSubTopic": "PUBSUB_TOPIC"
    }
  }
}

Dabei gilt:

PROJECT_ID: die ID des Google Cloud-Projekts, in dem Sie den Dataflow-Job ausführen möchten
JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
VERSION: die Version der Vorlage, die Sie verwenden möchten
Sie können die folgenden Werte verwenden:
- latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
- Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
Achtung: Die neueste Version der Vorlagen wird möglicherweise mit funktionsgefährdenden Änderungen aktualisiert. In Ihren Produktionsumgebungen sollten Vorlagen verwendet werden, die sich im aktuellsten datierten übergeordneten Ordner befinden, um zu verhindern, dass diese funktionsgefährdenden Änderungen Ihre Produktionsworkflows beeinträchtigen.
LOCATION: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
BIGTABLE_INSTANCE_ID: Ihre Bigtable-Instanz-ID.
BIGTABLE_TABLE_ID: Ihre Bigtable-Tabellen-ID.
BIGTABLE_APPLICATION_PROFILE_ID: Ihre Bigtable-Anwendungsprofil-ID
PUBSUB_TOPIC: der Name des Pub/Sub-Zielthemas

Nächste Schritte

Informationen zu Dataflow-Vorlagen
Sehen Sie sich die Liste der von Google bereitgestellten Vorlagen an.