Modello di flussi di modifiche di Spanner a Pub/Sub

Il modello di modifiche in tempo reale di Spanner in Pub/Sub è una pipeline di streaming che trasmette i record delle modifiche dei dati di Spanner e li scrive negli argomenti Pub/Sub utilizzando Dataflow Portable Runner.

Nota: il connettore di modifiche in tempo reale di Spanner non supporta l'arresto di un job. Per arrestare la pipeline, annulla il job anziché arrestarlo. L'arresto di un job con il connettore di modifiche in tempo reale di Spanner può causare il blocco del job in uno stato di arresto. Per saperne di più, consulta Arresto nella documentazione di Spanner.

Per inviare i dati a un nuovo argomento Pub/Sub, devi prima creare l'argomento. Dopo la creazione, Pub/Sub genera e collega automaticamente un abbonamento al nuovo argomento. Se provi a inviare dati a un argomento Pub/Sub inesistente, la pipeline Dataflow genera un'eccezione e si blocca perché tenta continuamente di stabilire una connessione.

Se l'argomento Pub/Sub necessario esiste già, puoi inviare i dati a quell'argomento.

Per saperne di più, consulta Informazioni sulle modifiche in tempo reale, Creare connessioni di modifiche in tempo reale con Dataflow, e Best practice per le modifiche in tempo reale.

Requisiti della pipeline

• L'istanza Spanner deve esistere prima dell'esecuzione della pipeline.
• Il database Spanner deve esistere prima dell'esecuzione della pipeline.
• L'istanza dei metadati di Spanner deve esistere prima dell'esecuzione della pipeline.
• Il database dei metadati di Spanner deve esistere prima dell'esecuzione della pipeline.
• La modifica in tempo reale di Spanner deve esistere prima dell'esecuzione della pipeline.
• L'argomento Pub/Sub deve esistere prima dell'esecuzione della pipeline.

Parametri del modello

Parametri obbligatori

• spannerInstanceId: l'istanza Spanner da cui leggere le modifiche in tempo reale.
• spannerDatabase: il database Spanner da cui leggere le modifiche in tempo reale.
• spannerMetadataInstanceId: l'istanza Spanner da utilizzare per la tabella dei metadati del connettore di modifiche in tempo reale.
• spannerMetadataDatabase: il database Spanner da utilizzare per la tabella dei metadati del connettore di modifiche in tempo reale.
• spannerChangeStreamName: il nome della modifica in tempo reale di Spanner da cui leggere.
• pubsubTopic: l'argomento Pub/Sub per l'output delle modifiche in tempo reale.

Parametri facoltativi

• spannerProjectId: il progetto da cui leggere le modifiche in tempo reale. In questo progetto viene creata anche la tabella dei metadati del connettore di modifiche in tempo reale. Il valore predefinito di questo parametro è il progetto in cui è in esecuzione la pipeline Dataflow.
• spannerDatabaseRole: il ruolo database di Spanner da utilizzare durante l'esecuzione del modello. Questo parametro è obbligatorio solo quando l'entità IAM che esegue il modello è un utente con controllo dell'accesso granulare. Il ruolo database deve avere il privilegio SELECT sulla modifica in tempo reale e il privilegio EXECUTE sulla funzione di lettura della modifica in tempo reale. Per saperne di più, consulta Controllo dell'accesso granulare per le modifiche in tempo reale (https://cloud.google.com/spanner/docs/fgac-change-streams).
• spannerMetadataTableName: il nome della tabella dei metadati del connettore di modifiche in tempo reale di Spanner da utilizzare. Se non viene fornito, Spanner crea automaticamente la tabella dei metadati del connettore di modifiche in tempo reale durante la modifica del flusso della pipeline. Devi fornire questo parametro quando aggiorni una pipeline esistente. Non utilizzare questo parametro in altri casi.
• startTimestamp: la data e l'ora di inizio (https://tools.ietf.org/html/rfc3339), inclusa, da utilizzare per la lettura delle modifiche in tempo reale. Ad esempio, 2021-10-12T07:20:50.52Z. Il valore predefinito è il timestamp di avvio della pipeline, ovvero l'ora corrente.
• endTimestamp: la data e l'ora di fine (https://tools.ietf.org/html/rfc3339), inclusa, da utilizzare per la lettura delle modifiche in tempo reale. Ad esempio, 2021-10-12T07:20:50.52Z. Il valore predefinito è un tempo infinito in futuro.
• spannerHost: l'endpoint Cloud Spanner da chiamare nel modello. Utilizzato solo per i test. Ad esempio, https://spanner.googleapis.com. Il valore predefinito è https://spanner.googleapis.com.
• outputDataFormat: il formato dell'output. L'output viene racchiuso in molti PubsubMessage e inviato a un argomento Pub/Sub. I formati consentiti sono JSON e AVRO. Il valore predefinito è JSON.
• pubsubAPI: l'API Pub/Sub utilizzata per implementare la pipeline. Le API consentite sono pubsubio e native_client. Per un numero ridotto di query al secondo (QPS), native_client ha una latenza inferiore. Per un numero elevato di QPS, pubsubio offre prestazioni migliori e più stabili. Il valore predefinito è pubsubio.
• pubsubProjectId: il progetto dell'argomento Pub/Sub. Il valore predefinito di questo parametro è il progetto in cui è in esecuzione la pipeline Dataflow.
• rpcPriority: la priorità della richiesta per le chiamate Spanner. I valori consentiti sono HIGH, MEDIUM e LOW. Il valore predefinito è HIGH.
• includeSpannerSource: indica se includere o meno l'ID del database Spanner e l'ID dell'istanza da cui leggere la modifica in tempo reale nei dati dei messaggi di output. Il valore predefinito è false.
• outputMessageMetadata: il valore stringa per il campo personalizzato outputMessageMetadata nel messaggio Pub/Sub di output. Il valore predefinito è vuoto e il campo outputMessageMetadata viene compilato solo se questo valore non è vuoto. Utilizza il carattere di escape per eventuali caratteri speciali quando inserisci il valore qui(ad es. virgolette doppie).
• useSpannerEmulatorHost: indica se utilizzare l'host Spanner configurato in getSpannerHost() come host dell'emulatore. Il valore predefinito è false.
• spannerChangeStreamTvfNameList: un elenco separato da punti e virgola dei nomi TVF di modifiche in tempo reale di Spanner da interrogare e unire. Il valore predefinito è vuoto.

Esegui il modello

Console

1. Vai alla pagina Dataflow Crea job da modello.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome job univoco.
3. (Facoltativo) In Endpoint regionale, seleziona un valore dal menu a discesa. La regione predefinita è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona il modello Modifiche in tempo reale di Cloud Spanner in Pub/Sub.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota: per utilizzare Google Cloud CLI per eseguire flex i modelli, devi avere la versione 284.0.0 o successiva di Google Cloud CLI.

Esegui il modello nella shell o nel terminale:

    gcloud dataflow flex-template run JOB_NAME \
        --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/ \
        --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
    

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale non datata nel bucket: gs://dataflow-templates-REGION_NAME/latest/
  • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella principale datata nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale datata più recente per evitare che queste modifiche che causano interruzioni influiscano sui workflow di produzione.
• REGION_NAME: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• SPANNER_INSTANCE_ID: l'ID istanza Spanner
• SPANNER_DATABASE: il database Spanner
• SPANNER_METADATA_INSTANCE_ID: l'ID istanza dei metadati di Spanner
• SPANNER_METADATA_DATABASE: il database dei metadati di Spanner
• SPANNER_CHANGE_STREAM: la modifica in tempo reale di Spanner
• PUBSUB_TOPIC: l'argomento Pub/Sub per l'output delle modifiche in tempo reale

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta HTTP POST. Per saperne di più sull' API e sui relativi ambiti di autorizzazione, 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/",
    }
  }
  

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto in cui vuoi eseguire il job Dataflow Google Cloud
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale non datata nel bucket: gs://dataflow-templates-REGION_NAME/latest/
  • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella principale datata nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale datata più recente per evitare che queste modifiche che causano interruzioni influiscano sui workflow di produzione.
• LOCATION: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• SPANNER_INSTANCE_ID: l'ID istanza Spanner
• SPANNER_DATABASE: il database Spanner
• SPANNER_METADATA_INSTANCE_ID: l'ID istanza dei metadati di Spanner
• SPANNER_METADATA_DATABASE: il database dei metadati di Spanner
• SPANNER_CHANGE_STREAM: la modifica in tempo reale di Spanner
• PUBSUB_TOPIC: l'argomento Pub/Sub per l'output delle modifiche in tempo reale

Codice sorgente del modello

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.Arrays;
import java.util.List;
import java.util.stream.Collectors;
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 = "googlecloud-to-googlecloud",
    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();

    String tvfNameListString = options.getSpannerChangeStreamTvfNameList();
    List<String> tvfNameList = null;
    if (tvfNameListString != null && !tvfNameListString.isEmpty()) {
      tvfNameList =
          Arrays.stream(tvfNameListString.split(";"))
              .filter(name -> !name.trim().isEmpty())
              .collect(Collectors.toList());
    }

    final RpcPriority rpcPriority = options.getRpcPriority();
    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withProjectId(spannerProjectId)
            .withInstanceId(instanceId)
            .withDatabaseId(databaseId);

    if (options.getUseSpannerEmulatorHost() != null && options.getUseSpannerEmulatorHost()) {
      spannerConfig =
          spannerConfig.withEmulatorHost(
              ValueProvider.StaticValueProvider.of(options.getSpannerHost()));
    } else {
      spannerConfig =
          spannerConfig.withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()));
    }

    // 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)
                .withTvfNameList(tvfNameList))
        .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();
  }
}

Passaggi successivi

• Scopri di più sui modelli Dataflow.
• Consulta l'elenco dei modelli forniti da Google.