Plantilla de cambios en tiempo real de Spanner a Cloud Storage

La plantilla de flujos de cambios de Spanner a Cloud Storage es un flujo de procesamiento en streaming que transmite registros de cambios de datos de Spanner y los escribe en un segmento de Cloud Storage mediante Dataflow Runner v2.

La canalización agrupa los registros de la secuencia de cambios de Spanner en ventanas según su marca de tiempo. Cada ventana representa un periodo de tiempo cuya duración puedes configurar con esta plantilla. Se garantiza que todos los registros con marcas de tiempo que pertenezcan a la ventana estarán en ella, por lo que no habrá llegadas tardías. También puedes definir un número de fragmentos de salida. La canalización crea un archivo de salida de Cloud Storage por ventana y por fragmento. Los registros de un archivo de salida no están ordenados. Los archivos de salida se pueden escribir en formato JSON o AVRO, según la configuración del usuario.

Ten en cuenta que puedes minimizar la latencia de la red y los costes de transporte de la red ejecutando la tarea de Dataflow desde la misma región que tu instancia de Spanner o tu segmento de Cloud Storage. Si usas fuentes, receptores, ubicaciones de archivos de almacenamiento temporal o ubicaciones de archivos temporales que se encuentran fuera de la región de tu trabajo, es posible que tus datos se envíen entre regiones. Consulta más información sobre las regiones de Dataflow.

Consulte más información sobre los flujos de cambios, cómo crear flujos de procesamiento de datos de flujos de cambios y las prácticas recomendadas.

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 segmento de salida de Cloud Storage debe existir antes de ejecutar el flujo de procesamiento.

Parámetros de plantilla

Parámetros obligatorios

• spannerInstanceId el ID de la instancia de Spanner desde la que se leerán los datos de los flujos de cambios.
• spannerDatabase la base de datos de Spanner de la que se van a leer los datos de los flujos de cambios.
• spannerMetadataInstanceId: el ID de instancia de Spanner que se va a usar en 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.
• gcsOutputDirectory la ruta y el prefijo de nombre de archivo para escribir archivos de salida. Debe terminar con una barra diagonal. El formato DateTime se usa para analizar la ruta del directorio de los formateadores de fecha y hora. Por ejemplo, gs://your-bucket/your-path.

Parámetros opcionales

• spannerProjectId el ID del proyecto de Google Cloud que contiene la base de datos de Spanner de la 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, se crea automáticamente una tabla de metadatos de cambio de Spanner durante la ejecución de la canalización. Debe proporcionar un valor para este parámetro al actualizar una canalización. De lo contrario, no lo uses.
• startTimestamp la fecha y hora de inicio (incluida) que se va a usar para leer los flujos de cambios, con el formato 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 de finalización (inclusive) que se va a 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.
• outputFileFormat: formato del archivo de Cloud Storage de salida. Los formatos admitidos son TEXT y AVRO. El valor predeterminado es AVRO.
• windowDuration es el intervalo en el que se escriben los datos en el directorio de salida. Configura la duración en función del rendimiento de la canalización. Por ejemplo, un mayor rendimiento puede requerir tamaños de ventana más pequeños para que los datos quepan en la memoria. El valor predeterminado es 5m (cinco minutos), con un mínimo de 1s (un segundo). Los formatos permitidos son: [int]s (segundos; por ejemplo, 5s), [int]m (minutos; por ejemplo, 12m) y [int]h (horas; por ejemplo, 2h). Por ejemplo, 5m.
• rpcPriority la prioridad de la solicitud para las llamadas de Spanner. El valor debe ser HIGH, MEDIUM o LOW. El valor predeterminado es HIGH.
• outputFilenamePrefix el prefijo que se colocará en cada archivo de ventana. Por ejemplo, output-. El valor predeterminado es "output".
• numShards: número máximo de fragmentos de salida que se producen al escribir. Un mayor número de particiones implica un mayor rendimiento para escribir en Cloud Storage, pero puede suponer un mayor coste de agregación de datos en las particiones al procesar los archivos de Cloud Storage de salida. El valor predeterminado es 20.

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 Google Cloud Storage 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_Google_Cloud_Storage \
    --region REGION_NAME \
    --parameters \
spannerInstanceId=SPANNER_INSTANCE_ID,\
spannerDatabase=SPANNER_DATABASE,\
spannerMetadataInstanceId=SPANNER_METADATA_INSTANCE_ID,\
spannerMetadataDatabase=SPANNER_METADATA_DATABASE,\
spannerChangeStreamName=SPANNER_CHANGE_STREAM,\
gcsOutputDirectory=GCS_OUTPUT_DIRECTORY

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 Cloud Spanner
• SPANNER_DATABASE: base de datos de Cloud Spanner
• SPANNER_METADATA_INSTANCE_ID: ID de instancia de metadatos de Cloud Spanner
• SPANNER_METADATA_DATABASE: base de datos de metadatos de Cloud Spanner
• SPANNER_CHANGE_STREAM: cambio en tiempo real de Cloud Spanner
• GCS_OUTPUT_DIRECTORY: ubicación del archivo de 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",
          "gcsOutputDirectory": "GCS_OUTPUT_DIRECTORY"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Spanner_Change_Streams_to_Google_Cloud_Storage",
   }
}

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 Cloud Spanner
• SPANNER_DATABASE: base de datos de Cloud Spanner
• SPANNER_METADATA_INSTANCE_ID: ID de instancia de metadatos de Cloud Spanner
• SPANNER_METADATA_DATABASE: base de datos de metadatos de Cloud Spanner
• SPANNER_CHANGE_STREAM: cambio en tiempo real de Cloud Spanner
• GCS_OUTPUT_DIRECTORY: ubicación del archivo de 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.SpannerChangeStreamsToGcsOptions;
import com.google.cloud.teleport.v2.transforms.FileFormatFactorySpannerChangeStreams;
import com.google.cloud.teleport.v2.utils.DurationUtils;
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.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link SpannerChangeStreamsToGcs} pipeline streams change stream record(s) and stores to
 * Google Cloud Storage bucket in user specified format. The sink data can be stored in a Text or
 * Avro file format.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-googlecloud/README_Spanner_Change_Streams_to_Google_Cloud_Storage.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Spanner_Change_Streams_to_Google_Cloud_Storage",
    category = TemplateCategory.STREAMING,
    displayName = "Cloud Spanner change streams to Cloud Storage",
    description = {
      "The Cloud Spanner change streams to Cloud Storage template is a streaming pipeline that streams Spanner data change records and writes them into a Cloud Storage bucket using Dataflow Runner V2.\n",
      "The pipeline groups Spanner change stream records into windows based on their timestamp, with each window representing a time duration whose length you can configure with this template. "
          + "All records with timestamps belonging to the window are guaranteed to be in the window; there can be no late arrivals. "
          + "You can also define a number of output shards; the pipeline creates one Cloud Storage output file per window per shard. "
          + "Within an output file, records are unordered. Output files can be written in either JSON or AVRO format, depending on the user configuration.\n",
      "Note that you can minimize network latency and network transport costs by running the Dataflow job from the same region as your Cloud Spanner instance or Cloud Storage bucket. "
          + "If you use sources, sinks, staging file locations, or temporary file locations that are located outside of your job's region, your data might be sent across regions. "
          + "See more about <a href=\"https://cloud.google.com/dataflow/docs/concepts/regional-endpoints\">Dataflow regional endpoints</a>.\n",
      "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 = SpannerChangeStreamsToGcsOptions.class,
    flexContainerName = "spanner-changestreams-to-gcs",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-spanner-change-streams-to-cloud-storage",
    contactInformation = "https://cloud.google.com/support",
    requirements = {
      "The Cloud Spanner instance must exist prior to 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 Cloud Storage output bucket must exist prior to running the pipeline."
    },
    streaming = true,
    supportsAtLeastOnce = true)
public class SpannerChangeStreamsToGcs {
  private static final Logger LOG = LoggerFactory.getLogger(SpannerChangeStreamsToGcs.class);
  private static final String USE_RUNNER_V2_EXPERIMENT = "use_runner_v2";

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

    LOG.info("Starting Input Files to GCS");

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

    run(options);
  }

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

  public static PipelineResult run(SpannerChangeStreamsToGcsOptions options) {
    LOG.info("Requested File Format is " + options.getOutputFileFormat());
    options.setStreaming(true);
    options.setEnableStreamingEngine(true);

    final Pipeline pipeline = Pipeline.create(options);

    // Get the Spanner project, instance, database, and change stream parameters.
    String projectId = getProjectId(options);
    String instanceId = options.getSpannerInstanceId();
    String databaseId = options.getSpannerDatabase();
    String metadataInstanceId = options.getSpannerMetadataInstanceId();
    String metadataDatabaseId = options.getSpannerMetadataDatabase();
    String changeStreamName = options.getSpannerChangeStreamName();

    // 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(projectId)
            .withInstanceId(instanceId)
            .withDatabaseId(databaseId);
    // Propagate database role for fine-grained access control on change stream.
    if (options.getSpannerDatabaseRole() != null) {
      LOG.info("Setting database role on SpannerConfig: " + options.getSpannerDatabaseRole());
      spannerConfig =
          spannerConfig.withDatabaseRole(
              ValueProvider.StaticValueProvider.of(options.getSpannerDatabaseRole()));
    }
    LOG.info("Created SpannerConfig: " + spannerConfig);
    pipeline
        .apply(
            SpannerIO.readChangeStream()
                .withSpannerConfig(spannerConfig)
                .withMetadataInstance(metadataInstanceId)
                .withMetadataDatabase(metadataDatabaseId)
                .withChangeStreamName(changeStreamName)
                .withInclusiveStartAt(startTimestamp)
                .withInclusiveEndAt(endTimestamp)
                .withRpcPriority(rpcPriority)
                .withMetadataTable(metadataTableName))
        .apply(
            "Creating " + options.getWindowDuration() + " Window",
            Window.into(FixedWindows.of(DurationUtils.parseDuration(options.getWindowDuration()))))
        .apply(
            "Write To GCS",
            FileFormatFactorySpannerChangeStreams.newBuilder().setOptions(options).build());

    return pipeline.run();
  }
}

Siguientes pasos

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