Plantilla de flujos de cambios de Bigtable a Pub/Sub

La plantilla de flujos de cambios de Bigtable a Pub/Sub es un flujo de procesamiento en streaming que transmite registros de cambios de datos de Bigtable y los publica en un tema de Pub/Sub mediante Dataflow.

Un flujo de cambios de Bigtable te permite suscribirte a mutaciones de datos por tabla. Cuando te suscribes a los flujos de cambios de una tabla, se aplican las siguientes restricciones:

• Solo se devuelven las celdas modificadas y los descriptores de las operaciones de eliminación.
• Solo se devuelve el nuevo valor de una celda modificada.

Cuando se publican registros de cambios de datos en un tema de Pub/Sub, los mensajes se pueden insertar en un orden diferente al orden original de las marcas de tiempo de confirmación de Bigtable.

Los registros de cambios de datos de Bigtable que no se pueden publicar en temas de Pub/Sub se colocan temporalmente en un directorio de cola de mensajes fallidos (cola de mensajes no procesados) en Cloud Storage. Una vez que se alcanza el número máximo de reintentos fallidos, estos registros se colocan indefinidamente en el mismo directorio de la cola de mensajes fallidos para que los revise un humano o para que el usuario los procese más adelante.

La canalización requiere que exista el tema de Pub/Sub de destino. El tema de destino puede configurarse para validar los mensajes mediante un esquema. Cuando un tema de Pub/Sub especifica un esquema, la canalización solo se inicia si el esquema es válido. En función del tipo de esquema, utilice una de las siguientes definiciones de esquema para el tema de destino:

Búferes de protocolo

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

Usa el siguiente esquema de Protobuf con la codificación de mensajes JSON:

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

Cada mensaje nuevo de Pub/Sub incluye una entrada de un registro de cambios de datos devuelto por el flujo de cambios de su fila correspondiente en tu tabla de Bigtable. La plantilla de Pub/Sub aplana las entradas de cada registro de cambio de datos en cambios individuales a nivel de celda.

Descripción del mensaje de salida de Pub/Sub

Nombre del campo	Descripción
rowKey	Clave de la fila modificada. Llega en forma de matriz de bytes. Cuando se configura la codificación de mensajes JSON, las claves de las filas se devuelven como cadenas. Cuando se especifica useBase64Rowkeys, las claves de las filas se codifican en Base64. De lo contrario, se usa un conjunto de caracteres especificado por bigtableChangeStreamCharset para decodificar los bytes de la clave de fila en una cadena.
modType	El tipo de mutación de la fila. Usa uno de los siguientes valores: SET_CELL, DELETE_CELLS o DELETE_FAMILY.
columnFamily	La familia de columnas afectada por la mutación de la fila.
column	El calificador de columna afectado por la mutación de la fila. En el caso del tipo de mutación DELETE_FAMILY, el campo de columna no está definido. Llega en forma de matriz de bytes. Cuando se configura la codificación de mensajes JSON, las columnas se devuelven como cadenas. Cuando se especifica useBase64ColumnQualifier, el campo de columna se codifica en Base64. De lo contrario, se usa un conjunto de caracteres especificado por bigtableChangeStreamCharset para decodificar los bytes de la clave de fila en una cadena.
commitTimestamp	La hora en la que Bigtable aplica la mutación. La hora se mide en microsegundos desde el inicio del registro de tiempo de Unix (1 de enero de 1970 a las 00:00:00 UTC).
timestamp	Valor de marca de tiempo de la celda afectada por la mutación. En los tipos de mutación DELETE_CELLS y DELETE_FAMILY, no se define ninguna marca de tiempo. La hora se mide en microsegundos desde el inicio del registro de tiempo de Unix (1 de enero de 1970 a las 00:00:00 UTC).
timestampFrom	Describe el inicio inclusivo del intervalo de marca de tiempo de todas las celdas eliminadas por la mutación DELETE_CELLS. En el caso de otros tipos de mutaciones, timestampFrom no se define. La hora se mide en microsegundos desde el inicio del registro de tiempo de Unix (1 de enero de 1970 a las 00:00:00 UTC).
timestampTo	Describe el final exclusivo del intervalo de marca de tiempo de todas las celdas eliminadas por la mutación DELETE_CELLS. En el caso de otros tipos de mutaciones, timestampTo no se define.
isGC	Valor booleano que indica si la mutación se ha generado mediante un mecanismo de recogida de elementos no utilizados de Bigtable.
tieBreaker	Cuando dos mutaciones se registran al mismo tiempo en diferentes clústeres de Bigtable, se aplica a la tabla de origen la mutación con el valor tiebreaker más alto. Las mutaciones con valores de tiebreaker más bajos se descartan.
value	El nuevo valor definido por la mutación. A menos que se defina la opción de canalización stripValues, el valor se asigna a las mutaciones de SET_CELL. En el caso de otros tipos de mutación, el valor no se define. Llega en forma de matriz de bytes. Cuando se configura la codificación de mensajes JSON, los valores se devuelven como cadenas. Si se especifica useBase64Values, el valor se codifica en Base64. De lo contrario, se usa un conjunto de caracteres especificado por bigtableChangeStreamCharset para decodificar los bytes de valor en una cadena.
sourceInstance	Nombre de la instancia de Bigtable que ha registrado la mutación. Puede ocurrir cuando varias canalizaciones transmiten cambios de diferentes instancias al mismo tema de Pub/Sub.
sourceCluster	El nombre del clúster de Bigtable que ha registrado la mutación. Se puede usar cuando varias canalizaciones transmiten cambios de diferentes instancias al mismo tema de Pub/Sub.
sourceTable	Nombre de la tabla de Bigtable que ha recibido la mutación. Se puede usar en el caso de que una transmisión de varias canalizaciones cambie de diferentes tablas al mismo tema de Pub/Sub.

Requisitos del flujo de procesamiento

• La instancia de origen de Bigtable especificada.
• La tabla de origen de Bigtable especificada. La tabla debe tener habilitados los flujos de cambios.
• El perfil de aplicación de Bigtable especificado.
• El tema de Pub/Sub especificado debe existir.

Parámetros de plantilla

Parámetros obligatorios

• pubSubTopic el nombre del tema de Pub/Sub de destino.
• bigtableChangeStreamAppProfile el ID del perfil de aplicación de Bigtable. El perfil de la aplicación debe usar el enrutamiento de un solo clúster y permitir transacciones de una sola fila.
• bigtableReadInstanceId el ID de la instancia de Bigtable de origen.
• bigtableReadTableId el ID de la tabla de Bigtable de origen.

Parámetros opcionales

• messageEncoding la codificación de los mensajes que se publicarán en el tema de Pub/Sub. Cuando se configura el esquema del tema de destino, la codificación de los mensajes se determina mediante los ajustes del tema. Se admiten los siguientes valores: BINARY y JSON. El valor predeterminado es JSON.
• messageFormat la codificación de los mensajes que se publicarán en el tema de Pub/Sub. Cuando se configura el esquema del tema de destino, la codificación de los mensajes se determina mediante los ajustes del tema. Se admiten los siguientes valores: AVRO, PROTOCOL_BUFFERS y JSON. El valor predeterminado es JSON. Cuando se usa el formato JSON, los campos rowKey, column y value del mensaje son cadenas cuyo contenido se determina mediante las opciones de la canalización useBase64Rowkeys, useBase64ColumnQualifiers, useBase64Values y bigtableChangeStreamCharset.
• stripValues si se define como true, las mutaciones de SET_CELL se devuelven sin que se hayan definido nuevos valores. El valor predeterminado es false. Este parámetro es útil cuando no necesitas que haya un valor nuevo (también conocido como invalidación de caché) o cuando los valores son extremadamente grandes y superan los límites de tamaño de los mensajes de Pub/Sub.
• dlqDirectory el directorio de la cola de mensajes fallidos. Los registros que no se pueden procesar se almacenan en este directorio. El valor predeterminado es un directorio de la ubicación temporal de la tarea de Dataflow. En la mayoría de los casos, puede usar la ruta predeterminada.
• dlqRetryMinutes el número de minutos que transcurren entre los reintentos de la cola de mensajes fallidos. El valor predeterminado es 10.
• dlqMaxRetries el número máximo de reintentos de mensajes fallidos. El valor predeterminado es 5.
• useBase64Rowkeys se usa con la codificación de mensajes JSON. Si se define como true, el campo rowKey es una cadena codificada en Base64. De lo contrario, rowKey se genera usando bigtableChangeStreamCharset para decodificar bytes en una cadena. El valor predeterminado es false.
• pubSubProjectId el ID del proyecto de Bigtable. El valor predeterminado es el proyecto de la tarea de Dataflow.
• useBase64ColumnQualifiers se usa con la codificación de mensajes JSON. Si se define como true, el campo column es una cadena codificada en Base64. De lo contrario, la columna se genera usando bigtableChangeStreamCharset para decodificar bytes en una cadena. El valor predeterminado es false.
• useBase64Values se usa con la codificación de mensajes JSON. Si se define como true, el campo de valor es una cadena codificada en Base64. De lo contrario, el valor se genera usando bigtableChangeStreamCharset para decodificar bytes en una cadena. El valor predeterminado es false.
• disableDlqRetries indica si se deben inhabilitar los reintentos de la cola de mensajes fallidos. Valor predeterminado: false.
• bigtableChangeStreamMetadataInstanceId el ID de instancia de metadatos de los flujos de cambios de Bigtable. El valor predeterminado es una cadena vacía.
• bigtableChangeStreamMetadataTableTableId el ID de la tabla de metadatos del conector de flujos de cambios de Bigtable. Si no se proporciona, se crea automáticamente una tabla de metadatos del conector de flujos de cambios de Bigtable durante la ejecución de la canalización. El valor predeterminado es una cadena vacía.
• bigtableChangeStreamCharset nombre del conjunto de caracteres de los flujos de cambios de Bigtable. El valor predeterminado es UTF-8.
• bigtableChangeStreamStartTimestamp marca de tiempo inicial (https://tools.ietf.org/html/rfc3339) (inclusive) que se usará para leer los flujos de cambios. Por ejemplo, 2022-05-05T07:59:59Z. El valor predeterminado es la marca de tiempo de la hora de inicio de la canalización.
• bigtableChangeStreamIgnoreColumnFamilies lista de cambios en los nombres de las familias de columnas separados por comas que se deben ignorar. El valor predeterminado es una cadena vacía.
• bigtableChangeStreamIgnoreColumns lista de cambios de nombres de columnas separados por comas que se deben ignorar. Ejemplo: "cf1:col1,cf2:col2". El valor predeterminado es una cadena vacía.
• bigtableChangeStreamName nombre único de la canalización del cliente. Te permite reanudar el procesamiento desde el punto en el que se detuvo una canalización que se estaba ejecutando. El valor predeterminado es un nombre generado automáticamente. Consulte los registros de trabajos de Dataflow para ver el valor utilizado.
• bigtableChangeStreamResume si se define como true, una nueva canalización reanuda el procesamiento desde el punto en el que se detuvo una canalización que se estaba ejecutando anteriormente con el mismo valor de bigtableChangeStreamName. Si la canalización con el valor bigtableChangeStreamName proporcionado no se ha ejecutado nunca, no se iniciará ninguna canalización. Si se define como false, se inicia una nueva canalización. Si ya se ha ejecutado una canalización con el mismo valor de bigtableChangeStreamName para la fuente en cuestión, no se iniciará una nueva canalización. El valor predeterminado es false.
• bigtableReadChangeStreamTimeoutMs tiempo de espera de las solicitudes ReadChangeStream de Bigtable en milisegundos.
• bigtableReadProjectId el ID del proyecto de Bigtable. El valor predeterminado es el proyecto de la tarea de Dataflow.

Ejecutar la plantilla

Consola

1. Ve a la página Crear tarea a partir de plantilla de Dataflow.
Ir a Crear tarea a partir de plantilla
2. En el campo Nombre de la tarea, introduce un nombre único.
3. Opcional: En Endpoint regional, seleccione un valor en el menú desplegable. La región predeterminada es us-central1.

   Para ver una lista de las regiones en las que puedes ejecutar una tarea de Dataflow, consulta Ubicaciones de Dataflow.

4. En el menú desplegable Plantilla de flujo de datos, seleccione the Bigtable change streams to Pub/Sub template.
5. En los campos de parámetros proporcionados, introduzca los valores de los parámetros.
6. Haz clic en Ejecutar trabajo.

gcloud

Nota: Para usar la CLI de Google Cloud y ejecutar plantillas de flex, debes tener la versión CLI de Google Cloud 284.0.0 o una posterior.

En tu shell o terminal, ejecuta la plantilla:

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

Haz los cambios siguientes:

• PROJECT_ID: el ID del proyecto Google Cloud en el que quieres ejecutar la tarea de Dataflow
• JOB_NAME: un nombre de trabajo único que elijas
• VERSION: la versión de la plantilla que quieres usar

  Puedes usar los siguientes valores:

  • latest para usar la última versión de la plantilla, que está disponible en la carpeta principal sin fecha del contenedor: gs://dataflow-templates-REGION_NAME/latest/
  • el nombre de la versión, como 2023-09-12-00_RC00, para usar una versión específica de la plantilla, que se encuentra anidada en la carpeta principal correspondiente con la fecha en el bucket: gs://dataflow-templates-REGION_NAME/
  Precaución: La última versión de las plantillas puede actualizarse con cambios importantes. Tus entornos de producción deben usar plantillas que se encuentren en la carpeta principal con fecha más reciente para evitar que estos cambios afecten a tus flujos de trabajo de producción.
• REGION_NAME: la región en la que quieras desplegar tu trabajo de Dataflow. Por ejemplo, us-central1
• BIGTABLE_INSTANCE_ID: el ID de tu instancia de Bigtable.
• BIGTABLE_TABLE_ID: el ID de tu tabla de Bigtable.
• BIGTABLE_APPLICATION_PROFILE_ID: el ID de tu perfil de aplicación de Bigtable.
• PUBSUB_TOPIC: el nombre del tema de destino de Pub/Sub

API

Para ejecutar la plantilla mediante la API REST, envía una solicitud HTTP POST. Para obtener más información sobre la API y sus ámbitos de autorización, consulta 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"
    }
  }
}

Haz los cambios siguientes:

• PROJECT_ID: el ID del proyecto Google Cloud en el que quieres ejecutar la tarea de Dataflow
• JOB_NAME: un nombre de trabajo único que elijas
• VERSION: la versión de la plantilla que quieres usar

  Puedes usar los siguientes valores:

  • latest para usar la última versión de la plantilla, que está disponible en la carpeta principal sin fecha del contenedor: gs://dataflow-templates-REGION_NAME/latest/
  • el nombre de la versión, como 2023-09-12-00_RC00, para usar una versión específica de la plantilla, que se encuentra anidada en la carpeta principal correspondiente con la fecha en el bucket: gs://dataflow-templates-REGION_NAME/
  Precaución: La última versión de las plantillas puede actualizarse con cambios importantes. Tus entornos de producción deben usar plantillas que se encuentren en la carpeta principal con fecha más reciente para evitar que estos cambios afecten a tus flujos de trabajo de producción.
• LOCATION: la región en la que quieras desplegar tu trabajo de Dataflow. Por ejemplo, us-central1
• BIGTABLE_INSTANCE_ID: el ID de tu instancia de Bigtable.
• BIGTABLE_TABLE_ID: el ID de tu tabla de Bigtable.
• BIGTABLE_APPLICATION_PROFILE_ID: el ID de tu perfil de aplicación de Bigtable.
• PUBSUB_TOPIC: el nombre del tema de destino de Pub/Sub

Código fuente de la plantilla

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.bigtablechangestreamstopubsub;

import com.google.cloud.Timestamp;
import com.google.cloud.bigtable.data.v2.models.ChangeStreamMutation;
import com.google.cloud.bigtable.data.v2.models.ChangeStreamMutation.MutationType;
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.pubsub.v1.SchemaServiceClient;
import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.bigtable.options.BigtableCommonOptions;
import com.google.cloud.teleport.v2.bigtable.utils.UnsupportedEntryException;
import com.google.cloud.teleport.v2.cdc.dlq.DeadLetterQueueManager;
import com.google.cloud.teleport.v2.cdc.dlq.StringDeadLetterQueueSanitizer;
import com.google.cloud.teleport.v2.coders.FailsafeElementCoder;
import com.google.cloud.teleport.v2.options.BigtableChangeStreamsToPubSubOptions;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.FailsafePublisher.PublishModJsonToTopic;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.model.MessageEncoding;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.model.MessageFormat;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.model.Mod;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.model.ModType;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.model.PubSubDestination;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.model.TestChangeStreamMutation;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstopubsub.schemautils.PubSubUtils;
import com.google.cloud.teleport.v2.transforms.DLQWriteTransform;
import com.google.cloud.teleport.v2.utils.BigtableSource;
import com.google.cloud.teleport.v2.values.FailsafeElement;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.Encoding;
import com.google.pubsub.v1.GetTopicRequest;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.Schema;
import com.google.pubsub.v1.SchemaName;
import com.google.pubsub.v1.Topic;
import com.google.pubsub.v1.TopicName;
import com.google.pubsub.v1.ValidateMessageRequest;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Base64;
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.coders.CoderRegistry;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.Flatten;
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.beam.sdk.values.PCollectionList;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.Duration;
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 PubsubMessage and
 * inserted into Pub/Sub topic.
 */
@Template(
    name = "Bigtable_Change_Streams_to_PubSub",
    category = TemplateCategory.STREAMING,
    displayName = "Cloud Bigtable Change Streams to PubSub",
    description =
        "Streaming pipeline. Streams Bigtable data change records and writes them into PubSub using Dataflow Runner V2.",
    optionsClass = BigtableChangeStreamsToPubSubOptions.class,
    optionsOrder = {
      BigtableChangeStreamsToPubSubOptions.class,
      BigtableCommonOptions.ReadChangeStreamOptions.class,
      BigtableCommonOptions.ReadOptions.class
    },
    skipOptions = {
      "bigtableReadAppProfile",
      "bigtableAdditionalRetryCodes",
      "bigtableRpcAttemptTimeoutMs",
      "bigtableRpcTimeoutMs"
    },
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-bigtable-change-streams-to-pubsub",
    flexContainerName = "bigtable-changestreams-to-pubsub",
    contactInformation = "https://cloud.google.com/support",
    streaming = true,
    supportsAtLeastOnce = true)
public final class BigtableChangeStreamsToPubSub {

  /** String/String Coder for {@link FailsafeElement}. */
  public static final FailsafeElementCoder<String, String> FAILSAFE_ELEMENT_CODER =
      FailsafeElementCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of());

  private static final Logger LOG = LoggerFactory.getLogger(BigtableChangeStreamsToPubSub.class);

  private static final String USE_RUNNER_V2_EXPERIMENT = "use_runner_v2";

  private static final TupleTag<FailsafeElement<String, String>> INVALID_MODS_TAG =
      new TupleTag<FailsafeElement<String, String>>("invalidMods") {};
  private static final TupleTag<FailsafeElement<String, String>> VALID_MODS_TAG =
      new TupleTag<FailsafeElement<String, String>>("validMods") {};

  /**
   * 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 PubSub");

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

    run(options);
  }

  private static void validateOptions(BigtableChangeStreamsToPubSubOptions options) {
    if (options.getDlqRetryMinutes() <= 0) {
      throw new IllegalArgumentException("dlqRetryMinutes must be positive.");
    }
    if (options.getDlqMaxRetries() < 0) {
      throw new IllegalArgumentException("dlqMaxRetries cannot be negative.");
    }
    if (options.getBigtableReadChangeStreamTimeoutMs() != null
        && options.getBigtableReadChangeStreamTimeoutMs() <= 0) {
      throw new IllegalArgumentException(
          "bigtableReadChangeStreamTimeoutMs must be greater than 0.");
    }
  }

  private static void setOptions(BigtableChangeStreamsToPubSubOptions 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(BigtableChangeStreamsToPubSubOptions options) {
    setOptions(options);
    validateOptions(options);

    String bigtableProject = getBigtableProjectId(options);

    // 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);

    Topic topic = null;
    try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
      GetTopicRequest request =
          GetTopicRequest.newBuilder()
              .setTopic(
                  TopicName.ofProjectTopicName(
                          getPubSubProjectId(options), options.getPubSubTopic())
                      .toString())
              .build();
      topic = topicAdminClient.getTopic(request);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }

    try {
      if (!validateSchema(topic, options, sourceInfo)) {
        final String errorMessage = "Configured topic doesn't accept messages of configured format";
        throw new IllegalArgumentException(errorMessage);
      }

    } catch (Exception e) {
      throw new IllegalArgumentException(e);
    }

    PubSubDestination destinationInfo = newPubSubDestination(options, topic);
    PubSubUtils pubSub = new PubSubUtils(sourceInfo, destinationInfo);

    /*
     * Stages:
     * 1) Read {@link ChangeStreamMutation} from change stream.
     * 2) Create {@link FailsafeElement} of {@link Mod} JSON and merge from:
     *    - {@link ChangeStreamMutation}.
     *    - GCS Dead letter queue.
     * 3) Convert {@link Mod} JSON into PubsubMessage and publish it to PubSub.
     * 4) Write Failures from 2) and 3) to GCS dead letter queue.
     */
    // Step 1
    Pipeline pipeline = Pipeline.create(options);

    // Register the coders for pipeline
    CoderRegistry coderRegistry = pipeline.getCoderRegistry();
    coderRegistry.registerCoderForType(
        FAILSAFE_ELEMENT_CODER.getEncodedTypeDescriptor(), FAILSAFE_ELEMENT_CODER);

    DeadLetterQueueManager dlqManager = buildDlqManager(options);

    String dlqDirectory = dlqManager.getRetryDlqDirectoryWithDateTime();
    String tempDlqDirectory = dlqManager.getRetryDlqDirectory() + "tmp/";

    if (options.getDisableDlqRetries()) {
      LOG.info(
          "Disabling retries for the DLQ, directly writing into severe DLQ: {}",
          dlqManager.getSevereDlqDirectoryWithDateTime());
      dlqDirectory = dlqManager.getSevereDlqDirectoryWithDateTime();
      tempDlqDirectory = dlqManager.getSevereDlqDirectory() + "tmp/";
    }

    BigtableIO.ReadChangeStream readChangeStream =
        BigtableIO.readChangeStream()
            .withChangeStreamName(options.getBigtableChangeStreamName())
            .withExistingPipelineOptions(
                options.getBigtableChangeStreamResume()
                    ? BigtableIO.ExistingPipelineOptions.RESUME_OR_FAIL
                    : BigtableIO.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());
    }

    if (options.getBigtableReadChangeStreamTimeoutMs() != null) {
      readChangeStream =
          readChangeStream.withReadChangeStreamTimeout(
              Duration.millis(options.getBigtableReadChangeStreamTimeoutMs()));
    }

    // Step 2: just return the output for sending to pubSub and DLQ
    PCollection<ChangeStreamMutation> dataChangeRecord =
        pipeline
            .apply("Read from Cloud Bigtable Change Streams", readChangeStream)
            .apply(Values.create());

    PCollection<FailsafeElement<String, String>> sourceFailsafeModJson =
        dataChangeRecord
            .apply(
                "ChangeStreamMutation To Mod JSON",
                ParDo.of(new ChangeStreamMutationToModJsonFn(sourceInfo)))
            .apply(
                "Wrap Mod JSON In FailsafeElement",
                ParDo.of(
                    new DoFn<String, FailsafeElement<String, String>>() {
                      @ProcessElement
                      public void process(
                          @Element String input,
                          OutputReceiver<FailsafeElement<String, String>> receiver) {
                        receiver.output(FailsafeElement.of(input, input));
                      }
                    }))
            .setCoder(FAILSAFE_ELEMENT_CODER);

    PCollectionTuple dlqModJson =
        dlqManager.getReconsumerDataTransform(
            pipeline.apply(dlqManager.dlqReconsumer(options.getDlqRetryMinutes())));

    PCollection<FailsafeElement<String, String>> retryableDlqFailsafeModJson = null;
    if (options.getDisableDlqRetries()) {
      retryableDlqFailsafeModJson = pipeline.apply(Create.empty(FAILSAFE_ELEMENT_CODER));
    } else {
      retryableDlqFailsafeModJson =
          dlqModJson.get(DeadLetterQueueManager.RETRYABLE_ERRORS).setCoder(FAILSAFE_ELEMENT_CODER);
    }

    PCollection<FailsafeElement<String, String>> failsafeModJson =
        PCollectionList.of(sourceFailsafeModJson)
            .and(retryableDlqFailsafeModJson)
            .apply("Merge Source And DLQ Mod JSON", Flatten.pCollections());

    PublishModJsonToTopic publishModJsonToTopic =
        new PublishModJsonToTopic(pubSub, VALID_MODS_TAG, INVALID_MODS_TAG);

    PCollectionTuple failedToPublish =
        failsafeModJson.apply("Publish Mod JSON To Pubsub", publishModJsonToTopic);

    PCollection<String> transformDlqJson =
        failedToPublish
            .get(VALID_MODS_TAG)
            .setCoder(FAILSAFE_ELEMENT_CODER)
            .apply(
                "Failed Mod JSON During Table Row Transformation",
                MapElements.via(new StringDeadLetterQueueSanitizer()));

    PCollectionList.of(transformDlqJson)
        .apply("Merge Failed Mod JSON From Transform And PubSub", Flatten.pCollections())
        .apply(
            "Write Failed Mod JSON To DLQ",
            DLQWriteTransform.WriteDLQ.newBuilder()
                .withDlqDirectory(dlqDirectory)
                .withTmpDirectory(tempDlqDirectory)
                .setIncludePaneInfo(true)
                .build());

    PCollection<FailsafeElement<String, String>> nonRetryableDlqModJsonFailsafe =
        PCollectionList.of(
                dlqModJson
                    .get(DeadLetterQueueManager.PERMANENT_ERRORS)
                    .setCoder(FAILSAFE_ELEMENT_CODER))
            .and(failedToPublish.get(INVALID_MODS_TAG).setCoder(FAILSAFE_ELEMENT_CODER))
            .apply("Merge Reconsume And Invalid Mods", Flatten.pCollections());
    LOG.info(
        "DLQ manager severe DLQ directory with date time: {}",
        dlqManager.getSevereDlqDirectoryWithDateTime());
    LOG.info("DLQ manager severe DLQ directory: {}", dlqManager.getSevereDlqDirectory() + "tmp/");
    nonRetryableDlqModJsonFailsafe
        .apply(
            "Write Mod JSON With Non-retriable Error To DLQ",
            MapElements.via(new StringDeadLetterQueueSanitizer()))
        .setCoder(StringUtf8Coder.of())
        .apply(
            DLQWriteTransform.WriteDLQ.newBuilder()
                .withDlqDirectory(dlqManager.getSevereDlqDirectoryWithDateTime())
                .withTmpDirectory(dlqManager.getSevereDlqDirectory() + "tmp/")
                .setIncludePaneInfo(true)
                .build());

    return pipeline.run();
  }

  private static PubSubDestination newPubSubDestination(
      BigtableChangeStreamsToPubSubOptions options, Topic topic) {
    return new PubSubDestination(
        getPubSubProjectId(options),
        options.getPubSubTopic(),
        topic,
        options.getMessageFormat(),
        options.getMessageEncoding(),
        options.getUseBase64Rowkeys(),
        options.getUseBase64ColumnQualifiers(),
        options.getUseBase64Values(),
        options.getStripValues());
  }

  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(
      BigtableChangeStreamsToPubSubOptions 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("DLQ directory: {}", dlqDirectory);
    return DeadLetterQueueManager.create(dlqDirectory, options.getDlqMaxRetries());
  }

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

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

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

  private static Boolean validateSchema(
      Topic topic, BigtableChangeStreamsToPubSubOptions options, BigtableSource source)
      throws Exception {
    String messageFormatPath = topic.getSchemaSettings().getSchema();
    if (topic.getSchemaSettings().getSchema().isEmpty()) {
      validateIncompatibleEncoding(options);
      LOG.info(
          "Topic has no schema configured, pipeline will use message format: {}, message encoding: {}",
          options.getMessageFormat(),
          options.getMessageEncoding());
      return true;
    } else {
      SchemaName schemaName = SchemaName.parse(topic.getSchemaSettings().getSchema());
      Schema schema;
      try (SchemaServiceClient schemaServiceClient = SchemaServiceClient.create()) {
        schema = schemaServiceClient.getSchema(schemaName);
      }

      options.setMessageEncoding(toMessageEncoding(topic.getSchemaSettings().getEncoding()));

      Schema.Type schemaType = schema.getType();
      switch (schemaType) {
        case AVRO:
          options.setMessageFormat(MessageFormat.AVRO);
          validateNoUseOfBase64(options);
          break;
        case PROTOCOL_BUFFER:
          if (options.getMessageEncoding() == MessageEncoding.JSON) {
            options.setMessageFormat(MessageFormat.JSON);
          } else {
            options.setMessageFormat(MessageFormat.PROTOCOL_BUFFERS);
            validateNoUseOfBase64(options);
          }
          break;
        case TYPE_UNSPECIFIED:
        case UNRECOGNIZED:
          // Not overriding messageFormat, will try what customer configured or the default if
          // not configured
          break;
        default:
          throw new IllegalArgumentException("Topic schema type is not supported: " + schemaType);
      }

      LOG.info("Topic has schema configured: {}", topic.getSchemaSettings().getSchema());
      LOG.info(
          "Pipeline will use message format: {}, message encoding: {}",
          options.getMessageFormat(),
          options.getMessageEncoding());

      PubSubDestination destination = newPubSubDestination(options, topic);
      PubSubUtils pubSub = new PubSubUtils(source, destination);

      ByteString testChangeMessageData = createTestChangeMessage(pubSub).getData();
      Encoding encoding = toPubSubEncoding(options.getMessageEncoding());
      try (SchemaServiceClient schemaServiceClient = SchemaServiceClient.create()) {
        String testMessageEncoded = toBase64String(testChangeMessageData);
        LOG.info("Validating a test message (Base64 encoded): {}", testMessageEncoded);
        ValidateMessageRequest request =
            ValidateMessageRequest.newBuilder()
                .setParent("projects/" + pubSub.getDestination().getPubSubProject())
                .setEncoding(encoding)
                .setMessage(testChangeMessageData)
                .setName(messageFormatPath)
                .build();
        schemaServiceClient.validateMessage(request);
        LOG.info("Test message successfully validated.");
      } catch (Exception e) {
        throw new IllegalArgumentException("Failed to validate test message", e);
      }
    }
    return true;
  }

  private static void validateNoUseOfBase64(BigtableChangeStreamsToPubSubOptions options) {
    if (options.getUseBase64Values()) {
      throw new IllegalArgumentException(
          "useBase64Values values can only be used with topics accepting JSON messages");
    }
    if (options.getUseBase64Rowkeys()) {
      throw new IllegalArgumentException(
          "useBase64Rowkeys values can only be used with topics accepting JSON messages");
    }
    if (options.getUseBase64ColumnQualifiers()) {
      throw new IllegalArgumentException(
          "useBase64ColumnQualifiers values can only be used with topics accepting JSON messages");
    }
  }

  private static void validateIncompatibleEncoding(BigtableChangeStreamsToPubSubOptions options) {
    if (options.getMessageEncoding() == MessageEncoding.BINARY
        && options.getMessageFormat() == MessageFormat.JSON) {
      throw new IllegalArgumentException(
          "JSON message format is incompatible with BINARY message encoding");
    }
  }

  private static MessageEncoding toMessageEncoding(Encoding encoding) {
    if (encoding == null) {
      return MessageEncoding.JSON;
    }
    switch (encoding) {
      case JSON:
      case ENCODING_UNSPECIFIED:
      case UNRECOGNIZED:
        return MessageEncoding.JSON;
      case BINARY:
        return MessageEncoding.BINARY;
      default:
        throw new IllegalArgumentException("Topic has unsupported message encoding: " + encoding);
    }
  }

  private static String toBase64String(ByteString testChangeMessageData) {
    return Base64.getEncoder().encodeToString(testChangeMessageData.toByteArray());
  }

  private static Encoding toPubSubEncoding(MessageEncoding messageEncoding) {
    switch (messageEncoding) {
      case BINARY:
        return Encoding.BINARY;
      case JSON:
        return Encoding.JSON;
      default:
        throw new IllegalArgumentException("Unexpected message encoding: " + messageEncoding);
    }
  }

  private static PubsubMessage createTestChangeMessage(PubSubUtils pubSub) throws Exception {
    SetCell setCell =
        SetCell.create(
            "test_column_family",
            ByteString.copyFrom("test_column", Charset.defaultCharset()),
            1000L, // timestamp
            ByteString.copyFrom("test_value", Charset.defaultCharset()));

    TestChangeStreamMutation mutation =
        new TestChangeStreamMutation(
            "test_rowkey",
            MutationType.USER,
            "source_cluster",
            java.time.Instant.now(), // commit timestamp
            1, // tiebreaker
            "token",
            java.time.Instant.now(), // low watermark
            setCell);

    Mod mod = new Mod(pubSub.getSource(), mutation, setCell);

    switch (pubSub.getDestination().getMessageFormat()) {
      case AVRO:
        return pubSub.mapChangeJsonStringToPubSubMessageAsAvro(mod.getChangeJson());
      case PROTOCOL_BUFFERS:
        return pubSub.mapChangeJsonStringToPubSubMessageAsProto(mod.getChangeJson());
      case JSON:
        return pubSub.mapChangeJsonStringToPubSubMessageAsJson(mod.getChangeJson());
      default:
        throw new IllegalArgumentException(
            "Unexpected message format: " + pubSub.getDestination().getMessageFormat());
    }
  }

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

    ChangeStreamMutationToModJsonFn(BigtableSource source) {
      this.sourceInfo = source;
    }

    private static String toJsonString(Mod mod, ChangeStreamMutation inputMutation) {
      try {
        return mod.toJson();
      } catch (IOException e) {
        // Ignore exception and print bad format.
        return String.format("\"%s\"", inputMutation);
      }
    }

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

        switch (modType) {
          case SET_CELL:
            SetCell setCell = (SetCell) entry;
            if (!sourceInfo.isIgnoredColumnFamily(setCell.getFamilyName())
                && !sourceInfo.isIgnoredColumn(
                    setCell.getFamilyName(),
                    setCell.getQualifier().toString(Charset.forName(sourceInfo.getCharset())))) {
              Mod mod = new Mod(sourceInfo, input, setCell);
              receiver.output(toJsonString(mod, input));
            }
            break;
          case DELETE_CELLS:
            DeleteCells deleteCells = (DeleteCells) entry;
            if (!sourceInfo.isIgnoredColumnFamily(deleteCells.getFamilyName())
                && !sourceInfo.isIgnoredColumn(
                    deleteCells.getFamilyName(),
                    deleteCells
                        .getQualifier()
                        .toString(Charset.forName(sourceInfo.getCharset())))) {
              Mod mod = new Mod(sourceInfo, input, deleteCells);
              receiver.output(toJsonString(mod, input));
            }
            break;
          case DELETE_FAMILY:
            DeleteFamily deleteFamily = (DeleteFamily) entry;
            if (!sourceInfo.isIgnoredColumnFamily(deleteFamily.getFamilyName())) {
              Mod mod = new Mod(sourceInfo, input, deleteFamily);
              receiver.output(toJsonString(mod, input));
            }
            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 DLQ directory. "
                    + "Please update your Dataflow template with the latest template version");
        }
      }
    }

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

Siguientes pasos

• Consulta información sobre las plantillas de Dataflow.
• Consulta la lista de plantillas proporcionadas por Google.