Vorlage „Bigtable-Änderungsstreams zu BigQuery“

Die Vorlage „Bigtable-Änderungsstreams zu BigQuery“ ist eine Streamingpipeline, die Bigtable-Datenänderungseinträge streamt und mithilfe von Dataflow in BigQuery-Tabellen schreibt.

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 Datenänderungsdatensätze in BigQuery geschrieben werden, können Zeilen in der falschen Reihenfolge eingefügt werden im Vergleich zur ursprünglichen Bigtable-Commit-Zeitstempelreihenfolge.

Änderungslog-Zeilen, die aufgrund eines persistenten Fehlers nicht in BigQuery geschrieben werden können, werden dauerhaft in ein Verzeichnis für unzustellbare Nachrichten (Cloud Storage für nicht verarbeitete Nachrichten) in Cloud Storage geschrieben, damit sie von Nutzern analysiert oder weiter verarbeitet werden können.

Wenn die erforderliche BigQuery-Tabelle nicht vorhanden ist, wird sie von der Pipeline erstellt. Andernfalls wird eine vorhandene BigQuery-Tabelle verwendet. Das Schema vorhandener BigQuery-Tabellen muss die Spalten in der folgenden Tabelle enthalten.

Jede neue BigQuery-Zeile enthält einen Datenänderungsdatensatz, der vom Änderungsstream aus der entsprechenden Zeile in Ihrer Bigtable-Tabelle zurückgegeben wird.

BigQuery-Ausgabetabellenschema

Spaltenname	Typ	Nullwerte zulässig	Beschreibung
row_key	STRING oder BYTES	Nein	Der Zeilenschlüssel der geänderten Zeile. Wenn die Pipelineoption writeRowkeyAsBytes auf true gesetzt ist, muss der Typ der Spalte BYTES sein. Andernfalls verwenden Sie den Typ STRING.
mod_type	STRING	Nein	Der Typ der Zeilenmutation. Verwenden Sie einen der folgenden Werte: SET_CELL, DELETE_CELLS oder DELETE_FAMILY.
column_family	STRING	Nein	Die von der Zeilenmutation betroffene Spaltenfamilie
column	STRING	Ja	Der von der Zeilenmutation betroffene Spaltenqualifizierer. Legen Sie für den Mutationstyp DELETE_FAMILY den Wert NULL fest.
commit_timestamp	TIMESTAMP	Nein	Der Zeitpunkt, an dem Bigtable die Mutation anwendet.
big_query_commit_timestamp	TIMESTAMP	Ja	Optional: Gibt die Zeit an, zu der BigQuery die Zeile in eine Ausgabetabelle schreibt. Das Feld wird nicht ausgefüllt, wenn der Spaltenname im Wert der Pipelineoption bigQueryChangelogTableFieldsToIgnore vorhanden ist.
timestamp	TIMESTAMP oder INT64	Ja	Der Zeitstempelwert der Zelle, die von der Mutation betroffen ist. Wenn die Pipelineoption writeNumericTimestamps auf true gesetzt ist, muss der Typ der Spalte INT64 sein. Andernfalls verwenden Sie den Typ TIMESTAMP. Legen Sie für die Mutationstypen DELETE_CELLS und DELETE_FAMILY den Wert NULL fest.
timestamp_from	TIMESTAMP oder INT64	Ja	Beschreibt einen inklusiven Start des Zeitstempelintervalls für alle Zellen, die von der Mutation DELETE_CELLS gelöscht wurden. Legen Sie für andere Mutationstypen NULL fest.
timestamp_to	TIMESTAMP oder INT64	Ja	Beschreibt ein exklusives Ende des Zeitstempelintervalls für alle Zellen, die mit der Mutation DELETE_CELLS gelöscht werden. Legen Sie für andere Mutationstypen NULL fest.
is_gc	BOOL	Nein	Optional: Wenn die Mutation durch eine Richtlinie für die automatische Speicherbereinigung ausgelöst wird, legen Sie true fest. In allen anderen Fällen legen Sie false fest. Das Feld wird nicht ausgefüllt, wenn der Spaltenname im Wert der Pipelineoption bigQueryChangelogTableFieldsToIgnore enthalten ist.
source_instance	STRING	Nein	Optional: Beschreibt den Namen der Bigtable-Instanz, von der die Mutation stammt. Das Feld wird nicht ausgefüllt, wenn der Spaltenname im Wert der Pipelineoption bigQueryChangelogTableFieldsToIgnore enthalten ist.
source_cluster	STRING	Nein	Optional: Beschreibt den Namen des Bigtable-Clusters, aus dem die Mutation stammt. Das Feld wird nicht ausgefüllt, wenn der Spaltenname im Wert der Pipelineoption bigQueryChangelogTableFieldsToIgnore enthalten ist.
source_table	STRING	Nein	Optional: Beschreibt den Namen der Bigtable-Tabelle, auf die die Mutation angewendet wird. Der Wert in dieser Spalte kann nützlich sein, wenn mehrere Bigtable-Tabellen Änderungen an derselben BigQuery-Tabelle streamen. Das Feld wird nicht ausgefüllt, wenn der Spaltenname im Wert der Pipelineoption bigQueryChangelogTableFieldsToIgnore enthalten ist.
tiebreaker	INT64	Nein	Optional: 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. Das Feld wird nicht ausgefüllt, wenn der Spaltenname im Wert der Pipelineoption bigQueryChangelogTableFieldsToIgnore enthalten ist.
value	STRING oder BYTES	Ja	Der neue Wert, der von der Mutation festgelegt wird. Wenn die Pipelineoption writeValuesAsBytes auf true gesetzt ist, muss der Typ der Spalte BYTES sein. Andernfalls verwenden Sie den Typ STRING. Der Wert wird für SET_CELL-Mutationen festgelegt. Bei anderen Mutationstypen wird der Wert auf NULL festgelegt.

Pipelineanforderungen

• Die angegebene Bigtable-Quellinstanz.
• Die angegebene Bigtable-Quelltabelle In der Tabelle müssen Änderungsstreams aktiviert sein.
• Das angegebene Bigtable-Anwendungsprofil.
• Das angegebene BigQuery-Ziel-Dataset.

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.
bigQueryDataset	Der Dataset-Name der BigQuery-Zieltabelle.
writeNumericTimestamps	Optional: Schreiben Sie den Bigtable-Zeitstempel als BigQuery-INT64. Wenn true festgelegt ist, werden Werte in die Spalte INT64 geschrieben. Andernfalls werden Werte in die Spalte TIMESTAMP geschrieben. Betroffene Spalten: timestamp, timestamp_from und timestamp_to Die Standardeinstellung ist false. Wenn dieser Wert auf true gesetzt ist, wird die Zeit seit der Unix-Epoche (1. Januar 1970 um UTC) in Mikrosekunden gemessen.
writeRowkeyAsBytes	Optional: Zeilenschlüssel als BigQuery-BYTES schreiben. Wenn true festgelegt ist, werden Zeilenschlüssel in die Spalte BYTES geschrieben. Andernfalls werden Zeilenschlüssel in die Spalte STRING geschrieben. Die Standardeinstellung ist false.
writeValuesAsBytes	Optional: Werte als BigQuery-BYTES schreiben. Wenn true festgelegt ist, werden Werte in die Spalte BYTES geschrieben. Andernfalls werden Werte in die Spalte STRING geschrieben. Die Standardeinstellung ist false.
bigQueryChangelogTableName	Optional: BigQuery-Zieltabellenname. Wenn nicht angegeben, wird der Wert bigtableReadTableId + "_changelog" verwendet.
bigQueryProjectId	Optional: Die Projekt-ID des BigQuery-Datasets. Der Standardwert ist das Projekt für den Dataflow-Job.
bigtableReadProjectId	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 Metadatentabellen-ID.
bigtableChangeStreamCharset	Optional: Die Bigtable-Änderung streamt den Namen des Zeichensatzes beim Lesen von Werten und Spaltenqualifizierern.
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. Weitere Informationen 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.
bigQueryChangelogTableFieldsToIgnore	Optional: Eine durch Kommas getrennte Liste der Änderungslogspalten, die, wenn angegeben, nicht erstellt und ausgefüllt werden. Verwenden Sie einen der folgenden unterstützten Werte: is_gc, source_instance, source_cluster, source_table, tiebreaker oder big_query_commit_timestamp. Standardmäßig werden alle Spalten ausgefüllt.
bigQueryChangelogTablePartitionExpirationMs	Optional: Legt die Ablaufzeit der Partition des Änderungslogs in Millisekunden fest. Wenn true festgelegt ist, werden Partitionen gelöscht, die älter als die angegebene Anzahl von Millisekunden sind. Standardmäßig ist kein Ablauf festgelegt.
bigQueryChangelogTablePartitionGranularity	Optional: Gibt einen Detaillierungsgrad zur Partitionierung der Änderungslogtabelle an. Wenn dieses Flag festgelegt ist, wird die Tabelle partitioniert. Verwenden Sie einen der folgenden unterstützten Werte: HOUR, DAY, MONTH oder YEAR. Standardmäßig ist die Tabelle nicht partitioniert.
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

1. Rufen Sie die Dataflow-Seite Job aus Vorlage erstellen auf.
Zur Seite "Job aus Vorlage erstellen“
2. Geben Sie im Feld Jobname einen eindeutigen Jobnamen ein.
3. 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.

4. Wählen Sie im Drop-down-Menü Dataflow-Vorlage die Option the Bigtable change streams to BigQuery template aus.
5. Geben Sie Ihre Parameterwerte in die Parameterfelder ein.
6. Klicken Sie auf Job ausführen.

gcloud

Hinweis: Um flex-Vorlagen mit dem Google Cloud CLI auszuführen, benötigen Sie das Google Cloud CLI in der Version 284.0.0 oder höher.

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_BigQuery \
    --parameters \
bigtableReadInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableReadTableId=BIGTABLE_TABLE_ID,\
bigtableChangeStreamAppProfile=BIGTABLE_APPLICATION_PROFILE_ID,\
bigQueryDataset=BIGQUERY_DESTINATION_DATASET

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
• BIGQUERY_DESTINATION_DATASET: der Name des BigQuery-Ziel-Datasets

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_BigQuery",
    "parameters": {
        "bigtableReadInstanceId": "BIGTABLE_INSTANCE_ID",
        "bigtableReadTableId": "BIGTABLE_TABLE_ID",
        "bigtableChangeStreamAppProfile": "BIGTABLE_APPLICATION_PROFILE_ID",
        "bigQueryDataset": "BIGQUERY_DESTINATION_DATASET"
    }
  }
}

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
• BIGQUERY_DESTINATION_DATASET: der Name des BigQuery-Ziel-Datasets

Quellcode der Vorlage

Java

/*
 * Copyright (C) 2023 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery;

import com.google.api.services.bigquery.model.TableRow;
import com.google.cloud.Timestamp;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.DatasetId;
import com.google.cloud.bigtable.data.v2.models.ChangeStreamMutation;
import com.google.cloud.bigtable.data.v2.models.DeleteCells;
import com.google.cloud.bigtable.data.v2.models.DeleteFamily;
import com.google.cloud.bigtable.data.v2.models.Entry;
import com.google.cloud.bigtable.data.v2.models.SetCell;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.bigtable.options.BigtableCommonOptions.ReadChangeStreamOptions;
import com.google.cloud.teleport.v2.bigtable.options.BigtableCommonOptions.ReadOptions;
import com.google.cloud.teleport.v2.bigtable.utils.UnsupportedEntryException;
import com.google.cloud.teleport.v2.cdc.dlq.DeadLetterQueueManager;
import com.google.cloud.teleport.v2.options.BigtableChangeStreamToBigQueryOptions;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.model.BigQueryDestination;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.model.BigtableSource;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.model.Mod;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.model.ModType;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.schemautils.BigQueryUtils;
import com.google.cloud.teleport.v2.transforms.DLQWriteTransform;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.InsertRetryPolicy;
import org.apache.beam.sdk.io.gcp.bigquery.WriteResult;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO.ExistingPipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Values;
import org.apache.beam.sdk.values.PCollection;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This pipeline ingests {@link ChangeStreamMutation} from Bigtable change stream. The {@link
 * ChangeStreamMutation} is then broken into {@link Mod}, which converted into {@link TableRow} and
 * inserted into BigQuery table.
 */
@Template(
    name = "Bigtable_Change_Streams_to_BigQuery",
    category = TemplateCategory.STREAMING,
    displayName = "Cloud Bigtable Change Streams to BigQuery",
    description =
        "Streaming pipeline. Streams Bigtable data change records and writes them into BigQuery using Dataflow Runner V2.",
    optionsClass = BigtableChangeStreamToBigQueryOptions.class,
    optionsOrder = {
      BigtableChangeStreamToBigQueryOptions.class,
      ReadChangeStreamOptions.class,
      ReadOptions.class
    },
    skipOptions = {
      "bigtableReadAppProfile",
      "bigtableAdditionalRetryCodes",
      "bigtableRpcAttemptTimeoutMs",
      "bigtableRpcTimeoutMs"
    },
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-bigtable-change-streams-to-bigquery",
    flexContainerName = "bigtable-changestreams-to-bigquery",
    contactInformation = "https://cloud.google.com/support",
    streaming = true)
public final class BigtableChangeStreamsToBigQuery {
  private static final Logger LOG = LoggerFactory.getLogger(BigtableChangeStreamsToBigQuery.class);

  private static final String USE_RUNNER_V2_EXPERIMENT = "use_runner_v2";

  /**
   * Main entry point for executing the pipeline.
   *
   * @param args The command-line arguments to the pipeline.
   */
  public static void main(String[] args) {
    LOG.info("Starting to replicate change records from Cloud Bigtable change streams to BigQuery");

    BigtableChangeStreamToBigQueryOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(BigtableChangeStreamToBigQueryOptions.class);

    run(options);
  }

  private static void setOptions(BigtableChangeStreamToBigQueryOptions options) {
    options.setStreaming(true);
    options.setEnableStreamingEngine(true);

    // Add use_runner_v2 to the experiments option, since change streams connector is only supported
    // on Dataflow runner v2.
    List<String> experiments = options.getExperiments();
    if (experiments == null) {
      experiments = new ArrayList<>();
    }
    boolean hasUseRunnerV2 = false;
    for (String experiment : experiments) {
      if (experiment.equalsIgnoreCase(USE_RUNNER_V2_EXPERIMENT)) {
        hasUseRunnerV2 = true;
        break;
      }
    }
    if (!hasUseRunnerV2) {
      experiments.add(USE_RUNNER_V2_EXPERIMENT);
    }
    options.setExperiments(experiments);
  }

  /**
   * Runs the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  public static PipelineResult run(BigtableChangeStreamToBigQueryOptions options) {
    setOptions(options);

    String changelogTableName = getBigQueryChangelogTableName(options);
    String bigtableProject = getBigtableProjectId(options);
    String bigQueryProject = getBigQueryProjectId(options);
    String bigQueryDataset = options.getBigQueryDataset();

    // If dataset doesn't exist and not checked, pipeline will start failing only after it sees the
    // first change from Cloud Bigtable. BigQueryIO can create table if it doesn't exist, but it
    // cannot create a dataset
    validateBigQueryDatasetExists(bigQueryProject, bigQueryDataset);

    // Retrieve and parse the startTimestamp
    Instant startTimestamp =
        options.getBigtableChangeStreamStartTimestamp().isEmpty()
            ? Instant.now()
            : toInstant(Timestamp.parseTimestamp(options.getBigtableChangeStreamStartTimestamp()));

    BigtableSource sourceInfo =
        new BigtableSource(
            options.getBigtableReadInstanceId(),
            options.getBigtableReadTableId(),
            getBigtableCharset(options),
            options.getBigtableChangeStreamIgnoreColumnFamilies(),
            options.getBigtableChangeStreamIgnoreColumns(),
            startTimestamp);

    BigQueryDestination destinationInfo =
        new BigQueryDestination(
            bigQueryProject,
            bigQueryDataset,
            changelogTableName,
            options.getWriteRowkeyAsBytes(),
            options.getWriteValuesAsBytes(),
            options.getWriteNumericTimestamps(),
            options.getBigQueryChangelogTablePartitionGranularity(),
            options.getBigQueryChangelogTablePartitionExpirationMs(),
            options.getBigQueryChangelogTableFieldsToIgnore());

    BigQueryUtils bigQuery = new BigQueryUtils(sourceInfo, destinationInfo);

    Pipeline pipeline = Pipeline.create(options);
    DeadLetterQueueManager dlqManager = buildDlqManager(options);

    BigtableIO.ReadChangeStream readChangeStream =
        BigtableIO.readChangeStream()
            .withChangeStreamName(options.getBigtableChangeStreamName())
            .withExistingPipelineOptions(
                options.getBigtableChangeStreamResume()
                    ? ExistingPipelineOptions.RESUME_OR_FAIL
                    : ExistingPipelineOptions.FAIL_IF_EXISTS)
            .withProjectId(bigtableProject)
            .withMetadataTableInstanceId(options.getBigtableChangeStreamMetadataInstanceId())
            .withInstanceId(options.getBigtableReadInstanceId())
            .withTableId(options.getBigtableReadTableId())
            .withAppProfileId(options.getBigtableChangeStreamAppProfile())
            .withStartTime(startTimestamp);

    if (!StringUtils.isBlank(options.getBigtableChangeStreamMetadataTableTableId())) {
      readChangeStream =
          readChangeStream.withMetadataTableTableId(
              options.getBigtableChangeStreamMetadataTableTableId());
    }

    PCollection<ChangeStreamMutation> dataChangeRecord =
        pipeline
            .apply("Read from Cloud Bigtable Change Streams", readChangeStream)
            .apply(Values.create());

    PCollection<TableRow> changeStreamMutationToTableRow =
        dataChangeRecord.apply(
            "ChangeStreamMutation To TableRow",
            ParDo.of(new ChangeStreamMutationToTableRowFn(sourceInfo, bigQuery)));

    Write<TableRow> bigQueryWrite =
        BigQueryIO.<TableRow>write()
            .to(destinationInfo.getBigQueryTableReference())
            .withSchema(bigQuery.getDestinationTableSchema())
            .withFormatFunction(element -> element)
            .withFormatRecordOnFailureFunction(element -> element)
            .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
            .withWriteDisposition(WriteDisposition.WRITE_APPEND)
            .withExtendedErrorInfo()
            .withMethod(Write.Method.STORAGE_API_AT_LEAST_ONCE)
            .withNumStorageWriteApiStreams(0)
            .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors());

    if (destinationInfo.isPartitioned()) {
      bigQueryWrite = bigQueryWrite.withTimePartitioning(bigQuery.getTimePartitioning());
    }

    // Unfortunately, due to https://github.com/apache/beam/issues/24090, it is no longer possible
    // to pass metadata via fake columns when writing to BigQuery. Previously we'd pass something
    // like retry count and then format it out before writing, but BQ would return original object
    // which would allow us to increment retry count and store it to DLQ with incremented number.
    // Because WRITE API doesn't allow access to original object, all metadata values are stripped
    // and we can only rely on retry policy and put all other persistently failing rows to DLQ as
    // a non-retriable severe failure.
    //
    // Since we're not going to be retrying such failures, we'll not use any reading from DLQ
    // capability.

    WriteResult writeResult =
        changeStreamMutationToTableRow.apply("Write To BigQuery", bigQueryWrite);

    writeResult
        .getFailedStorageApiInserts()
        .apply(
            "Failed Mod JSON During BigQuery Writes",
            MapElements.via(new BigQueryDeadLetterQueueSanitizer()))
        .apply(
            "Write rejected TableRow JSON To DLQ",
            DLQWriteTransform.WriteDLQ.newBuilder()
                .withDlqDirectory(dlqManager.getSevereDlqDirectory() + "YYYY/MM/dd/HH/mm/")
                .withTmpDirectory(dlqManager.getSevereDlqDirectory() + "tmp/")
                .setIncludePaneInfo(true)
                .build());

    return pipeline.run();
  }

  private static void validateBigQueryDatasetExists(
      String bigQueryProject, String bigQueryDataset) {
    BigQueryOptions options = BigQueryOptions.newBuilder().build();
    options.setThrowNotFound(true);

    BigQuery bigQuery = options.getService();
    bigQuery.getDataset(DatasetId.of(bigQueryProject, bigQueryDataset));
  }

  private static Instant toInstant(Timestamp timestamp) {
    if (timestamp == null) {
      return null;
    } else {
      return Instant.ofEpochMilli(timestamp.getSeconds() * 1000 + timestamp.getNanos() / 1000000);
    }
  }

  private static DeadLetterQueueManager buildDlqManager(
      BigtableChangeStreamToBigQueryOptions options) {
    String tempLocation =
        options.as(DataflowPipelineOptions.class).getTempLocation().endsWith("/")
            ? options.as(DataflowPipelineOptions.class).getTempLocation()
            : options.as(DataflowPipelineOptions.class).getTempLocation() + "/";
    String dlqDirectory =
        options.getDlqDirectory().isEmpty() ? tempLocation + "dlq/" : options.getDlqDirectory();

    LOG.info("Dead letter queue directory: {}", dlqDirectory);
    return DeadLetterQueueManager.create(dlqDirectory, 1);
  }

  private static String getBigtableCharset(BigtableChangeStreamToBigQueryOptions options) {
    return StringUtils.isEmpty(options.getBigtableChangeStreamCharset())
        ? "UTF-8"
        : options.getBigtableChangeStreamCharset();
  }

  private static String getBigtableProjectId(BigtableChangeStreamToBigQueryOptions options) {
    return StringUtils.isEmpty(options.getBigtableReadProjectId())
        ? options.getProject()
        : options.getBigtableReadProjectId();
  }

  private static String getBigQueryChangelogTableName(
      BigtableChangeStreamToBigQueryOptions options) {
    return StringUtils.isEmpty(options.getBigQueryChangelogTableName())
        ? options.getBigtableReadTableId() + "_changelog"
        : options.getBigQueryChangelogTableName();
  }

  private static String getBigQueryProjectId(BigtableChangeStreamToBigQueryOptions options) {
    return StringUtils.isEmpty(options.getBigQueryProjectId())
        ? options.getProject()
        : options.getBigQueryProjectId();
  }

  /**
   * DoFn that converts a {@link ChangeStreamMutation} to multiple {@link Mod} in serialized JSON
   * format.
   */
  static class ChangeStreamMutationToTableRowFn extends DoFn<ChangeStreamMutation, TableRow> {
    private final BigtableSource sourceInfo;
    private final BigQueryUtils bigQuery;

    ChangeStreamMutationToTableRowFn(BigtableSource source, BigQueryUtils bigQuery) {
      this.sourceInfo = source;
      this.bigQuery = bigQuery;
    }

    @ProcessElement
    public void process(@Element ChangeStreamMutation input, OutputReceiver<TableRow> receiver)
        throws Exception {
      for (Entry entry : input.getEntries()) {
        ModType modType = getModType(entry);

        Mod mod = null;
        switch (modType) {
          case SET_CELL:
            mod = new Mod(sourceInfo, input, (SetCell) entry);
            break;
          case DELETE_CELLS:
            mod = new Mod(sourceInfo, input, (DeleteCells) entry);
            break;
          case DELETE_FAMILY:
            mod = new Mod(sourceInfo, input, (DeleteFamily) entry);
            break;
          default:
          case UNKNOWN:
            throw new UnsupportedEntryException(
                "Cloud Bigtable change stream entry of type "
                    + entry.getClass().getName()
                    + " is not supported. The entry was put into a dead letter queue directory. "
                    + "Please update your Dataflow template with the latest template version");
        }

        TableRow tableRow = new TableRow();
        if (bigQuery.setTableRowFields(mod, tableRow)) {
          receiver.output(tableRow);
        }
      }
    }

    private ModType getModType(Entry entry) {
      if (entry instanceof SetCell) {
        return ModType.SET_CELL;
      } else if (entry instanceof DeleteCells) {
        return ModType.DELETE_CELLS;
      } else if (entry instanceof DeleteFamily) {
        return ModType.DELETE_FAMILY;
      }
      return ModType.UNKNOWN;
    }
  }
}

Nächste Schritte

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