Modelo de fluxos de alterações do Spanner para Pub/Sub

O modelo de fluxos de alterações do Spanner para o Pub/Sub é um pipeline de streaming que transmite os registros de alteração de dados do Spanner e os grava em tópicos do Pub/Sub usando o Dataflow Runner V2.

Você precisa criar o novo tópico do Pub/Sub antes de enviar seus dados. Depois de criar, o Pub/Sub gera e anexa automaticamente uma assinatura ao novo tópico. Se você tentar enviar dados para um tópico do Pub/Sub que não existe, o pipeline do Dataflow vai gerar uma exceção e o pipeline vai travar enquanto tenta fazer uma conexão.

Se o tópico do Pub/Sub necessário já existir, os dados podem ser gerados para ele.

Para mais informações, consulte Sobre fluxos de alteração, Criar conexões de fluxos de alteração com o Dataflow e Práticas recomendadas para fluxos de alteração.

Requisitos de pipeline

• A instância do Spanner precisa existir antes da execução do pipeline.
• O banco de dados do Spanner precisa ser criado antes da execução do pipeline.
• A instância de metadados do Spanner precisa existir antes da execução do pipeline.
• O banco de dados de metadados do Spanner precisa existir antes da execução do pipeline.
• O fluxo de alterações do Spanner precisa ser criado antes da execução do pipeline.
• O tópico do Pub/Sub precisa ser criado antes da execução do pipeline.

Parâmetros do modelo

Parâmetros obrigatórios

• spannerInstanceId: a instância do Spanner em que os fluxos de alterações serão lidos.
• spannerDatabase: o banco de dados do Spanner de onde os fluxos de alterações serão lidos.
• spannerMetadataInstanceId: a instância do Spanner a ser usada para a tabela de metadados do conector dos fluxos de alterações.
• spannerMetadataDatabase: o banco de dados do Spanner a ser usado para a tabela de metadados do conector dos fluxos de alterações.
• spannerChangeStreamName: o nome do fluxo de alterações do Spanner a ser lido.
• pubsubTopic: o tópico do Pub/Sub para saída do fluxo de alterações.

Parâmetros opcionais

• spannerProjectId: o projeto do qual os fluxos de alterações serão lidos. Esse projeto também é onde a tabela de metadados do conector de fluxos de alteração é criada. O padrão para esse parâmetro é o projeto em que o pipeline do Dataflow está sendo executado.
• spannerDatabaseRole: o papel do banco de dados do Spanner a ser usado ao executar o modelo. Esse parâmetro é necessário somente quando o principal do IAM que executa o modelo é um usuário de controle de acesso minucioso. A função de banco de dados precisa ter o privilégio SELECT no fluxo de alterações e o privilégio EXECUTE na função de leitura do fluxo de alterações. Para mais informações, consulte "Controle de acesso granular para fluxos de alteração" (https://cloud.google.com/spanner/docs/fgac-change-streams).
• spannerMetadataTableName: o nome da tabela de metadados do conector dos fluxos de alterações do Spanner a ser usado. Se não for informado, o Spanner automaticamente criará a tabela de metadados do conector dos fluxos durante a mudança do fluxo do pipeline. Você precisa fornecer esse parâmetro ao atualizar um pipeline atual. Não use esse parâmetro para outros casos.
• startTimestamp: o DateTime inicial (https://tools.ietf.org/html/rfc3339), inclusivo, a ser usado na leitura de fluxos de alterações. Por exemplo, ex- 2021-10-12T07:20:50.52Z. O padrão é o carimbo de data/hora em que o pipeline é iniciado, ou seja, o horário atual.
• endTimestamp: o DateTime final (https://tools.ietf.org/html/rfc3339), inclusivo, a ser usado na leitura de fluxos de alterações. Por exemplo, ex- 2021-10-12T07:20:50.52Z. O padrão é um tempo infinito no futuro.
• spannerHost: o endpoint do Cloud Spanner a ser chamado no modelo. Usado apenas para testes. Por exemplo, https://spanner.googleapis.com. O padrão é: https://spanner.googleapis.com.
• outputDataFormat: o formato da saída. A saída é encapsulada em muitas PubsubMessages e enviada para um tópico do Pub/Sub. Os formatos permitidos são JSON e AVRO. O padrão é JSON.
• pubsubAPI: a API Pub/Sub usada para implementar o pipeline. As APIs permitidas são pubsubio e native_client. Para um pequeno número de consultas por segundo (QPS), native_client tem menos latência. Quando o QPS é alto, pubsubio tem um desempenho melhor e mais estável. O padrão é pubsubio.
• pubsubProjectId: projeto do tópico do Pub/Sub. O padrão para esse parâmetro é o projeto em que o pipeline do Dataflow está sendo executado.
• rpcPriority: a prioridade de solicitação das chamadas do Spanner. Os valores permitidos são ALTO, MÉDIO e BAIXO. O valor padrão é ALTO.
• includeSpannerSource: indica se o ID do banco de dados do Spanner e o ID da instância devem ser incluídos para ler o fluxo de alterações nos dados da mensagem de saída. O padrão é: falso.
• outputMessageMetadata: o valor da string para o campo personalizado outputMessageMetadata na mensagem de saída do Pub/Sub. O padrão é vazio, e o campo outputMessageMetadata só é preenchido se esse valor não estiver vazio. Use caracteres de escape para qualquer caractere especial ao inserir o valor aqui(por exemplo, aspas duplas).

Executar o modelo

Console

1. Acesse a página Criar job usando um modelo do Dataflow.
Acesse Criar job usando um modelo
2. No campo Nome do job, insira um nome exclusivo.
3. Opcional: em Endpoint regional, selecione um valor no menu suspenso. A região padrão é us-central1.

   Para ver uma lista de regiões em que é possível executar um job do Dataflow, consulte Locais do Dataflow.

4. No menu suspenso Modelo do Dataflow, selecione the Cloud Spanner change streams to Pub/Sub template.
5. Nos campos de parâmetro fornecidos, insira os valores de parâmetro.
6. Cliquem em Executar job.

gcloud

Observação: para usar a Google Cloud CLI para executar modelos flex você precisa ter a Google Cloud CLI na versão 284.0.0 ou mais recente.

No shell ou no terminal, execute o modelo:

    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
    

Substitua:

• JOB_NAME: um nome de job de sua escolha
• VERSION: a versão do modelo que você quer usar

  Use estes valores:

  • latest para usar a versão mais recente do modelo, disponível na pasta mãe não datada no bucket: gs://dataflow-templates-REGION_NAME/latest/
  • o nome da versão, como 2023-09-12-00_RC00, para usar uma versão específica do modelo, que pode ser encontrada aninhada na respectiva pasta mãe datada no bucket: gs://dataflow-templates-REGION_NAME/
  Cuidado: é possível que a versão mais recente dos modelos seja atualizada com alterações interruptivas. Os ambientes de produção precisam usar os modelos mantidos na pasta mãe datada mais recente para evitar que essas alterações interruptivas afetem os fluxos de trabalho de produção.
• REGION_NAME: a região em que você quer implantar o job do Dataflow, por exemplo, us-central1
• SPANNER_INSTANCE_ID: ID da instância do Spanner
• SPANNER_DATABASE: banco de dados do Spanner
• SPANNER_METADATA_INSTANCE_ID: ID da instância de metadados do Spanner
• SPANNER_METADATA_DATABASE: banco de dados de metadados do Spanner
• SPANNER_CHANGE_STREAM: fluxo de alterações do Spanner
• PUBSUB_TOPIC: o tópico do Pub/Sub para saída dos fluxos de alteração

API

Para executar o modelo usando a API REST, envie uma solicitação HTTP POST. Para mais informações sobre a API e os respectivos escopos de autorização, consulte 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",
    }
  }
  

Substitua:

• PROJECT_ID: o ID do projeto do Google Cloud em que você quer executar o job do Dataflow
• JOB_NAME: um nome de job de sua escolha
• VERSION: a versão do modelo que você quer usar

  Use estes valores:

  • latest para usar a versão mais recente do modelo, disponível na pasta mãe não datada no bucket: gs://dataflow-templates-REGION_NAME/latest/
  • o nome da versão, como 2023-09-12-00_RC00, para usar uma versão específica do modelo, que pode ser encontrada aninhada na respectiva pasta mãe datada no bucket: gs://dataflow-templates-REGION_NAME/
  Cuidado: é possível que a versão mais recente dos modelos seja atualizada com alterações interruptivas. Os ambientes de produção precisam usar os modelos mantidos na pasta mãe datada mais recente para evitar que essas alterações interruptivas afetem os fluxos de trabalho de produção.
• LOCATION: a região em que você quer implantar o job do Dataflow, por exemplo, us-central1
• SPANNER_INSTANCE_ID: ID da instância do Spanner
• SPANNER_DATABASE: banco de dados do Spanner
• SPANNER_METADATA_INSTANCE_ID: ID da instância de metadados do Spanner
• SPANNER_METADATA_DATABASE: banco de dados de metadados do Spanner
• SPANNER_CHANGE_STREAM: fluxo de alterações do Spanner
• PUBSUB_TOPIC: o tópico do Pub/Sub para saída dos fluxos de alteração

Código-fonte do modelo

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

A seguir

• Saiba mais sobre os modelos do Dataflow.
• Confira a lista de modelos fornecidos pelo Google.