Plantilla de flujos de cambios de Spanner a Pub/Sub

La plantilla de flujos de cambios de Spanner a Pub/Sub es un flujo de procesamiento en streaming que transmite registros de cambios de datos de Spanner y los escribe en temas de Pub/Sub con Dataflow Runner V2.

Para enviar sus datos a un nuevo tema de Pub/Sub, primero debe crear el tema. Después de crear un tema, Pub/Sub genera y adjunta automáticamente una suscripción al tema nuevo. Si intentas enviar datos a un tema de Pub/Sub que no existe, la canalización de flujo de datos genera una excepción y se bloquea, ya que intenta continuamente establecer una conexión.

Si ya existe el tema de Pub/Sub necesario, puedes enviar datos a ese tema.

Para obtener más información, consulta los artículos Acerca de los flujos de cambios, Crear conexiones de flujos de cambios con Dataflow y Prácticas recomendadas para usar flujos de cambios.

Requisitos del flujo de procesamiento

• La instancia de Spanner debe existir antes de ejecutar el flujo de procesamiento.
• La base de datos de Spanner debe existir antes de ejecutar el flujo de procesamiento.
• La instancia de metadatos de Spanner debe existir antes de ejecutar el flujo de procesamiento.
• La base de datos de metadatos de Spanner debe existir antes de ejecutar el flujo de procesamiento.
• El flujo de cambios de Spanner debe existir antes de ejecutar el flujo de procesamiento.
• El tema de Pub/Sub debe existir antes de ejecutar el flujo de procesamiento.

Parámetros de plantilla

Parámetros obligatorios

• spannerInstanceId la instancia de Spanner de la que se van a leer los flujos de cambios.
• spannerDatabase la base de datos de Spanner de la que se van a leer los flujos de cambios.
• spannerMetadataInstanceId: la instancia de Spanner que se va a usar para la tabla de metadatos del conector de flujos de cambios.
• spannerMetadataDatabase la base de datos de Spanner que se va a usar para la tabla de metadatos del conector de flujos de cambios.
• spannerChangeStreamName el nombre del flujo de cambios de Spanner del que se van a leer los datos.
• pubsubTopic el tema de Pub/Sub para la salida de los flujos de cambios.

Parámetros opcionales

• spannerProjectId el proyecto del que se van a leer los flujos de cambios. En este proyecto también se crea la tabla de metadatos del conector de flujos de cambios. El valor predeterminado de este parámetro es el proyecto en el que se ejecuta la canalización de Dataflow.
• spannerDatabaseRole el rol de base de datos de Spanner que se usará al ejecutar la plantilla. Este parámetro solo es obligatorio cuando la entidad de IAM que ejecuta la plantilla es un usuario con control de acceso detallado. El rol de la base de datos debe tener el privilegio SELECT en el flujo de cambios y el privilegio EXECUTE en la función de lectura del flujo de cambios. Para obtener más información, consulta Control de acceso granular para secuencias de cambios (https://cloud.google.com/spanner/docs/fgac-change-streams).
• spannerMetadataTableName nombre de la tabla de metadatos del conector de cambios en tiempo real de Spanner que se va a usar. Si no se proporciona, Spanner crea automáticamente la tabla de metadatos del conector de Streams durante el cambio de flujo de la canalización. Debe proporcionar este parámetro al actualizar una canalización. No utilice este parámetro en otros casos.
• startTimestamp la fecha y hora de inicio (https://tools.ietf.org/html/rfc3339), incluida, que se usará para leer los flujos de cambios. Por ejemplo, ex- 2021-10-12T07:20:50.52Z. El valor predeterminado es la marca de tiempo en la que se inicia la canalización, es decir, la hora actual.
• endTimestamp DateTime final (https://tools.ietf.org/html/rfc3339), incluido, que se usará para leer los flujos de cambios. Por ejemplo, ex- 2021-10-12T07:20:50.52Z. El valor predeterminado es un tiempo infinito en el futuro.
• spannerHost el endpoint de Cloud Spanner al que se llama en la plantilla. Solo se usa para hacer pruebas. Por ejemplo, https://spanner.googleapis.com. El valor predeterminado es https://spanner.googleapis.com.
• outputDataFormat el formato de la salida. La salida se envuelve en muchos PubsubMessages y se envía a un tema de Pub/Sub. Los formatos permitidos son JSON y AVRO. El valor predeterminado es JSON.
• pubsubAPI la API de Pub/Sub que se usa para implementar la canalización. Las APIs permitidas son pubsubio y native_client. Si el número de consultas por segundo (CPS) es bajo, native_client tiene menos latencia. Si el número de CPS es elevado, pubsubio ofrece un rendimiento mejor y más estable. El valor predeterminado es pubsubio.
• pubsubProjectId: proyecto del tema de Pub/Sub. El valor predeterminado de este parámetro es el proyecto en el que se ejecuta la canalización de Dataflow.
• rpcPriority la prioridad de la solicitud para las llamadas de Spanner. Los valores permitidos son HIGH, MEDIUM y LOW. El valor predeterminado es HIGH).
• includeSpannerSource indica si se debe incluir o no el ID de la base de datos y el ID de la instancia de Spanner para leer el flujo de cambios en los datos del mensaje de salida. Valor predeterminado: false.
• outputMessageMetadata valor de cadena del campo personalizado outputMessageMetadata en el mensaje de publicación/suscripción de salida. El valor predeterminado es una cadena vacía y el campo outputMessageMetadata solo se rellena si este valor no está vacío. Evite los caracteres especiales al introducir el valor(por ejemplo, las comillas dobles).

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 Cloud Spanner 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 \
        --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Spanner_Change_Streams_to_PubSub \
        --region REGION_NAME \
        --parameters \
    spannerInstanceId=SPANNER_INSTANCE_ID,\
    spannerDatabase=SPANNER_DATABASE,\
    spannerMetadataInstanceId=SPANNER_METADATA_INSTANCE_ID,\
    spannerMetadataDatabase=SPANNER_METADATA_DATABASE,\
    spannerChangeStreamName=SPANNER_CHANGE_STREAM,\
    pubsubTopic=PUBSUB_TOPIC
    

Haz los cambios siguientes:

• 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
• SPANNER_INSTANCE_ID: ID de instancia de Spanner
• SPANNER_DATABASE: base de datos de Spanner
• SPANNER_METADATA_INSTANCE_ID: ID de instancia de metadatos de Spanner
• SPANNER_METADATA_DATABASE: base de datos de metadatos de Spanner
• SPANNER_CHANGE_STREAM: flujo de cambios de Spanner
• PUBSUB_TOPIC: el tema de Pub/Sub para la salida de los flujos de cambios.

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",
        "parameters": {
            "spannerInstanceId": "SPANNER_INSTANCE_ID",
            "spannerDatabase": "SPANNER_DATABASE",
            "spannerMetadataInstanceId": "SPANNER_METADATA_INSTANCE_ID",
            "spannerMetadataDatabase": "SPANNER_METADATA_DATABASE",
            "spannerChangeStreamName": "SPANNER_CHANGE_STREAM",
            "pubsubTopic": "PUBSUB_TOPIC"
        },
        "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Spanner_Change_Streams_to_PubSub",
    }
  }
  

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
• SPANNER_INSTANCE_ID: ID de instancia de Spanner
• SPANNER_DATABASE: base de datos de Spanner
• SPANNER_METADATA_INSTANCE_ID: ID de instancia de metadatos de Spanner
• SPANNER_METADATA_DATABASE: base de datos de metadatos de Spanner
• SPANNER_CHANGE_STREAM: flujo de cambios de Spanner
• PUBSUB_TOPIC: el tema de Pub/Sub para la salida de los flujos de cambios.

Código fuente de la plantilla

Java

/*
 * Copyright (C) 2022 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;

import com.google.cloud.Timestamp;
import com.google.cloud.spanner.Options.RpcPriority;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.options.SpannerChangeStreamsToPubSubOptions;
import com.google.cloud.teleport.v2.transforms.FileFormatFactorySpannerChangeStreamsToPubSub;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link SpannerChangeStreamsToPubSub} pipeline streams change stream record(s) and stores to
 * pubsub topic in user specified format. The sink data can be stored in a JSON Text or Avro data
 * format.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-googlecloud/README_Spanner_Change_Streams_to_PubSub.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Spanner_Change_Streams_to_PubSub",
    category = TemplateCategory.STREAMING,
    displayName = "Cloud Spanner change streams to Pub/Sub",
    description = {
      "The Cloud Spanner change streams to the Pub/Sub template is a streaming pipeline that streams Cloud Spanner data change records and writes them into Pub/Sub topics using Dataflow Runner V2.\n",
      "To output your data to a new Pub/Sub topic, you need to first create the topic. After creation, Pub/Sub automatically generates and attaches a subscription to the new topic. "
          + "If you try to output data to a Pub/Sub topic that doesn't exist, the dataflow pipeline throws an exception, and the pipeline gets stuck as it continuously tries to make a connection.\n",
      "If the necessary Pub/Sub topic already exists, you can output data to that topic.",
      "Learn more about <a href=\"https://cloud.google.com/spanner/docs/change-streams\">change streams</a>, <a href=\"https://cloud.google.com/spanner/docs/change-streams/use-dataflow\">how to build change streams Dataflow pipelines</a>, and <a href=\"https://cloud.google.com/spanner/docs/change-streams/use-dataflow#best_practices\">best practices</a>."
    },
    optionsClass = SpannerChangeStreamsToPubSubOptions.class,
    flexContainerName = "spanner-changestreams-to-pubsub",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-spanner-change-streams-to-pubsub",
    contactInformation = "https://cloud.google.com/support",
    requirements = {
      "The Cloud Spanner instance must exist before running the pipeline.",
      "The Cloud Spanner database must exist prior to running the pipeline.",
      "The Cloud Spanner metadata instance must exist prior to running the pipeline.",
      "The Cloud Spanner metadata database must exist prior to running the pipeline.",
      "The Cloud Spanner change stream must exist prior to running the pipeline.",
      "The Pub/Sub topic must exist prior to running the pipeline."
    },
    streaming = true,
    supportsAtLeastOnce = true)
public class SpannerChangeStreamsToPubSub {
  private static final Logger LOG = LoggerFactory.getLogger(SpannerChangeStreamsToPubSub.class);
  private static final String USE_RUNNER_V2_EXPERIMENT = "use_runner_v2";

  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    LOG.info("Starting Input Messages to Pub/Sub");

    SpannerChangeStreamsToPubSubOptions options =
        PipelineOptionsFactory.fromArgs(args).as(SpannerChangeStreamsToPubSubOptions.class);

    run(options);
  }

  private static String getSpannerProjectId(SpannerChangeStreamsToPubSubOptions options) {
    return options.getSpannerProjectId().isEmpty()
        ? options.getProject()
        : options.getSpannerProjectId();
  }

  private static String getPubsubProjectId(SpannerChangeStreamsToPubSubOptions options) {
    return options.getPubsubProjectId().isEmpty()
        ? options.getProject()
        : options.getPubsubProjectId();
  }

  public static boolean isValidAsciiString(String outputMessageMetadata) {
    if (outputMessageMetadata != null
        && !StandardCharsets.US_ASCII.newEncoder().canEncode(outputMessageMetadata)) {
      return false;
    }
    return true;
  }

  public static PipelineResult run(SpannerChangeStreamsToPubSubOptions options) {
    LOG.info("Requested Message Format is " + options.getOutputDataFormat());
    options.setStreaming(true);
    options.setEnableStreamingEngine(true);

    final Pipeline pipeline = Pipeline.create(options);
    // Get the Spanner project, instance, database, metadata instance, metadata database
    // change stream, pubsub topic, and pubsub api parameters.
    String spannerProjectId = getSpannerProjectId(options);
    String instanceId = options.getSpannerInstanceId();
    String databaseId = options.getSpannerDatabase();
    String metadataInstanceId = options.getSpannerMetadataInstanceId();
    String metadataDatabaseId = options.getSpannerMetadataDatabase();
    String changeStreamName = options.getSpannerChangeStreamName();
    String pubsubProjectId = getPubsubProjectId(options);
    String pubsubTopicName = options.getPubsubTopic();
    String pubsubAPI = options.getPubsubAPI();
    Boolean includeSpannerSource = options.getIncludeSpannerSource();
    String outputMessageMetadata = options.getOutputMessageMetadata();

    // Ensure outputMessageMetadata only contains valid ascii characters
    if (!isValidAsciiString(outputMessageMetadata)) {
      throw new RuntimeException("outputMessageMetadata contains non ascii characters.");
    }

    // Retrieve and parse the start / end timestamps.
    Timestamp startTimestamp =
        options.getStartTimestamp().isEmpty()
            ? Timestamp.now()
            : Timestamp.parseTimestamp(options.getStartTimestamp());
    Timestamp endTimestamp =
        options.getEndTimestamp().isEmpty()
            ? Timestamp.MAX_VALUE
            : Timestamp.parseTimestamp(options.getEndTimestamp());

    // 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<>();
    }
    if (!experiments.contains(USE_RUNNER_V2_EXPERIMENT)) {
      experiments.add(USE_RUNNER_V2_EXPERIMENT);
    }
    options.setExperiments(experiments);

    String metadataTableName =
        options.getSpannerMetadataTableName() == null
            ? null
            : options.getSpannerMetadataTableName();

    final RpcPriority rpcPriority = options.getRpcPriority();
    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()))
            .withProjectId(spannerProjectId)
            .withInstanceId(instanceId)
            .withDatabaseId(databaseId);
    // Propagate database role for fine-grained access control on change stream.
    if (options.getSpannerDatabaseRole() != null) {
      spannerConfig =
          spannerConfig.withDatabaseRole(
              ValueProvider.StaticValueProvider.of(options.getSpannerDatabaseRole()));
    }
    pipeline
        .apply(
            SpannerIO.readChangeStream()
                .withSpannerConfig(spannerConfig)
                .withMetadataInstance(metadataInstanceId)
                .withMetadataDatabase(metadataDatabaseId)
                .withChangeStreamName(changeStreamName)
                .withInclusiveStartAt(startTimestamp)
                .withInclusiveEndAt(endTimestamp)
                .withRpcPriority(rpcPriority)
                .withMetadataTable(metadataTableName))
        .apply(
            "Convert each record to a PubsubMessage",
            FileFormatFactorySpannerChangeStreamsToPubSub.newBuilder()
                .setOutputDataFormat(options.getOutputDataFormat())
                .setProjectId(pubsubProjectId)
                .setPubsubAPI(pubsubAPI)
                .setPubsubTopicName(pubsubTopicName)
                .setIncludeSpannerSource(includeSpannerSource)
                .setSpannerDatabaseId(databaseId)
                .setSpannerInstanceId(instanceId)
                .setOutputMessageMetadata(outputMessageMetadata)
                .build());
    return pipeline.run();
  }
}

Siguientes pasos

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