Modello di flussi di modifiche di Spanner in Cloud Storage

Il modello di modifiche in tempo reale di Spanner in Cloud Storage è una pipeline di streaming che esegue lo streaming dei record delle modifiche dei dati di Spanner e li scrive in un bucket Cloud Storage 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.

La pipeline raggruppa i record delle modifiche in tempo reale di Spanner in finestre in base al relativo timestamp, con ogni finestra che rappresenta una durata di tempo la cui lunghezza può essere configurata con questo modello. Tutti i record con timestamp appartenenti alla finestra sono garantiti nella finestra; non possono esserci arrivi in ritardo. Puoi anche definire un numero di shard di output; la pipeline crea un file di output di Cloud Storage per finestra per shard. All'interno di un file di output, i record non sono ordinati. I file di output possono essere scritti in formato JSON o AVRO, a seconda della configurazione dell'utente.

Tieni presente che puoi ridurre al minimo la latenza di rete e i costi di trasporto di rete eseguendo il job Dataflow dalla stessa regione dell'istanza Spanner o del bucket Cloud Storage. Se utilizzi origini, sink, località dei file di staging o località dei file temporanei che si trovano al di fuori della regione del job, i dati potrebbero essere inviati tra le regioni. Scopri di più sulle regioni di Dataflow.

Scopri di più sulle modifiche in tempo reale, su come creare pipeline Dataflow di modifiche in tempo reale e sulle best practice.

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.
• Il bucket di output di Cloud Storage deve esistere prima dell'esecuzione della pipeline.

Parametri del modello

Parametri obbligatori

• spannerInstanceId: l'ID dell'istanza Spanner da cui leggere i dati delle modifiche in tempo reale.
• spannerDatabase: il database Spanner da cui leggere i dati delle modifiche in tempo reale.
• spannerMetadataInstanceId: l'ID dell'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.
• gcsOutputDirectory: il percorso e il prefisso del nome file per la scrittura dei file di output. Deve terminare con una barra. La formattazione DateTime viene utilizzata per analizzare il percorso della directory per i formattatori di data e ora. Ad esempio, gs://your-bucket/your-path.

Parametri facoltativi

• spannerProjectId: l'ID del progetto Google Cloud che contiene il database Spanner 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 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 fornita, durante l'esecuzione della pipeline viene creata automaticamente una tabella dei metadati delle modifiche in tempo reale di Spanner. Devi fornire un valore per questo parametro quando aggiorni una pipeline esistente. In caso contrario, non utilizzare questo parametro.
• startTimestamp: la data e l'ora di inizio, inclusiva, da utilizzare per la lettura delle modifiche in tempo reale, nel formato Ex-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, inclusiva, da utilizzare per la lettura delle modifiche in tempo reale. Ad esempio, Ex-2021-10-12T07:20:50.52Z. Il valore predefinito è un tempo infinito nel 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.
• outputFileFormat: il formato del file di output di Cloud Storage. I formati consentiti sono TEXT e AVRO. Il valore predefinito è AVRO.
• windowDuration: la durata della finestra è l'intervallo in cui i dati vengono scritti nella directory di output. Configura la durata in base alla velocità effettiva della pipeline. Ad esempio, una velocità effettiva più elevata potrebbe richiedere dimensioni delle finestre più piccole in modo che i dati rientrino nella memoria. Il valore predefinito è 5 m (cinque minuti), con un minimo di 1 s (un secondo). I formati consentiti sono: [int]s (per i secondi, ad esempio 5s), [int]m (per i minuti, ad esempio 12m), [int]h (per le ore, ad esempio 2h). Ad esempio, 5m.
• rpcPriority: la priorità della richiesta per le chiamate Spanner. Il valore deve essere HIGH, MEDIUM o LOW. Il valore predefinito è HIGH.
• spannerChangeStreamTvfNameList: elenco separato da punti e virgola dei nomi TVF di modifiche in tempo reale di Spanner da interrogare e unire. Il valore predefinito è vuoto.
• outputFilenamePrefix: il prefisso da inserire in ogni file con finestra. Ad esempio, output-. Il valore predefinito è: output.
• numShards: il numero massimo di shard di output prodotti durante la scrittura. Un numero maggiore di shard significa una velocità effettiva più elevata per la scrittura in Cloud Storage, ma potenzialmente un costo di aggregazione dei dati più elevato tra gli shard durante l'elaborazione dei file di output di Cloud Storage. Il valore predefinito è: 20.

Esegui il modello

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome job univoco.
3. (Facoltativo) Per Endpoint a livello di regione, 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. Nel menu a discesa Modello Dataflow, seleziona il modello Modifiche in tempo reale di Cloud Spanner in Google Cloud Storage.
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.

Nella shell o nel terminale, esegui il modello:

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

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: ID istanza Cloud Spanner
• SPANNER_DATABASE: database Cloud Spanner
• SPANNER_METADATA_INSTANCE_ID: ID istanza dei metadati di Cloud Spanner
• SPANNER_METADATA_DATABASE: database dei metadati di Cloud Spanner
• SPANNER_CHANGE_STREAM: modifica in tempo reale di Cloud Spanner
• GCS_OUTPUT_DIRECTORY: percorso del file 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",
          "gcsOutputDirectory": "GCS_OUTPUT_DIRECTORY"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Spanner_Change_Streams_to_Google_Cloud_Storage",
   }
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto Google Cloud in cui vuoi eseguire il job Dataflow
• 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: ID istanza Cloud Spanner
• SPANNER_DATABASE: database Cloud Spanner
• SPANNER_METADATA_INSTANCE_ID: ID istanza dei metadati di Cloud Spanner
• SPANNER_METADATA_DATABASE: database dei metadati di Cloud Spanner
• SPANNER_CHANGE_STREAM: modifica in tempo reale di Cloud Spanner
• GCS_OUTPUT_DIRECTORY: percorso del file 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.SpannerChangeStreamsToGcsOptions;
import com.google.cloud.teleport.v2.transforms.FileFormatFactorySpannerChangeStreams;
import com.google.cloud.teleport.v2.utils.DurationUtils;
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.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 = "googlecloud-to-googlecloud",
    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();

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

Passaggi successivi

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