Spanner change streams to Cloud Storage 模板

Spanner change streams to Cloud Storage 模板是一种流处理流水线，可使用 Dataflow Runner v2 流式传输 Spanner 数据更改记录并将其写入 Cloud Storage 存储桶。

流水线根据 Spanner 变更数据流记录的时间戳将其分组到窗口中，每个窗口代表一个时长，您可以使用此模板配置该时长。时间戳属于某个窗口的所有记录都保证在该窗口中；不会有延迟到达。您还可以定义多个输出分片。流水线会为每个窗口的每个分片创建一个 Cloud Storage 输出文件。在输出文件中，记录是无序的。输出文件可以采用 JSON 或 AVRO 格式编写，具体取决于用户配置。

请注意，通过在与 Spanner 实例或 Cloud Storage 存储桶相同的区域运行 Dataflow 作业，您可以最大限度地减少网络延迟和网络传输费用。如果您使用位于作业区域之外的源、接收器、暂存文件位置或临时文件位置，则数据可能会跨区域发送。详细了解 Dataflow 区域。

详细了解变更数据流、如何构建变更数据流 Dataflow 流水线和最佳实践。

流水线要求

  • 在运行流水线之前，Spanner 实例必须已存在。
  • 在运行流水线之前，Spanner 数据库必须已存在。
  • 在运行流水线之前，Spanner 元数据实例必须已存在。
  • 在运行流水线之前，Spanner 元数据数据库必须已存在。
  • 在运行流水线之前，Spanner 变更数据流必须已存在。
  • 在运行流水线之前，Cloud Storage 输出存储桶必须已存在。

模板参数

必需参数

  • spannerInstanceId：要从中读取变更数据流数据的 Spanner 实例 ID。
  • spannerDatabase：要从中读取变更数据流数据的 Spanner 数据库。
  • spannerMetadataInstanceId：要用于变更数据流连接器元数据表的 Spanner 实例 ID。
  • spannerMetadataDatabase：要用于变更数据流连接器元数据表的 Spanner 数据库。
  • spannerChangeStreamName：要从中读取数据的 Spanner 变更数据流的名称。
  • gcsOutputDirectory：用于写入输出文件的路径和文件名前缀。必须以斜杠结尾。 日期时间格式用于解析日期和时间格式化程序的目录路径。例如 gs://your-bucket/your-path。

可选参数

  • spannerProjectId：包含要从中读取变更数据流的 Spanner 数据库的 Google Cloud 项目的 ID。此项目也是创建变更数据流连接器元数据表的项目。此参数的默认项目是 Dataflow 流水线在其中运行的项目。
  • spannerDatabaseRole：运行模板时使用的 Spanner 数据库角色。仅当运行模板的 IAM 主账号是精细访问权限控制用户时，才需要此参数。数据库角色必须拥有变更数据流的 SELECT 特权和变更数据流的读取函数的 EXECUTE 特权。如需了解详情，请参阅变更数据流的精细访问权限控制 (https://cloud.google.com/spanner/docs/fgac-change-streams)。
  • spannerMetadataTableName：要使用的 Spanner 变更数据流连接器元数据表名称。如果未提供，系统会在流水线执行期间自动创建 Spanner 变更数据流元数据表。更新现有流水线时，您必须为此参数提供一个值。否则，请勿使用此参数。
  • startTimestamp：用于读取变更数据流的起始日期时间（含边界值），格式为 Ex-2021-10-12T07:20:50.52Z。默认为流水线启动时的时间戳，即当前时间。
  • endTimestamp：要用于读取变更数据流的结束日期时间（含边界值）。例如 Ex-2021-10-12T07:20:50.52Z。默认为未来的无限时间。
  • spannerHost：要在模板中调用的 Cloud Spanner 端点。仅用于测试。例如 https://spanner.googleapis.com。默认值为 https://spanner.googleapis.com。
  • outputFileFormat：输出 Cloud Storage 文件的格式。允许的格式为 TEXT 和 AVRO。默认值为 AVRO。
  • windowDuration：窗口时长是将数据写入输出目录的时间间隔。请根据流水线的吞吐量配置时长。例如，较高的吞吐量可能需要较短的窗口时长，以便数据适应内存。默认值为 5m（5 分钟），最少为 1s（1 秒）。允许的格式为 [int]（以秒为单位，例如 5s）、[int]m（以分钟为单位，例如 12m）、[int]h（以小时为单位，例如 2h）。 例如 5m。
  • rpcPriority：Spanner 调用的请求优先级。值必须为 HIGH、MEDIUM 或 LOW。默认值为 HIGH。
  • outputFilenamePrefix：要在各窗口文件上放置的前缀。例如 output-。默认值：output。
  • numShards：写入时生成的输出分片数上限。分片数越多，写入 Cloud Storage 的吞吐量越高，但处理输出 Cloud Storage 文件时跨分片聚合数据的费用也可能更高。默认值为 20。

运行模板

控制台gcloudAPI
  1. 转到 Dataflow 基于模板创建作业页面。
    2. 转到“基于模板创建作业”
  2. 在作业名称字段中，输入唯一的作业名称。
  3. 可选：对于区域性端点，从下拉菜单中选择一个值。默认区域为 us-central1。

    如需查看可以在其中运行 Dataflow 作业的区域列表，请参阅 Dataflow 位置。

  4. 从 Dataflow 模板下拉菜单中，选择 the Cloud Spanner change streams to Google Cloud Storage template。
  5. 在提供的参数字段中，输入您的参数值。
  6. 点击运行作业。

在 shell 或终端中，运行模板：

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

替换以下内容：

  • JOB_NAME：您选择的唯一性作业名称
  • VERSION：您要使用的模板的版本

    您可使用以下值：

  • REGION_NAME：要在其中部署 Dataflow 作业的区域，例如 us-central1
  • SPANNER_INSTANCE_ID：Cloud Spanner 实例 ID
  • SPANNER_DATABASE：Cloud Spanner 数据库
  • SPANNER_METADATA_INSTANCE_ID：Cloud Spanner 元数据实例 ID
  • SPANNER_METADATA_DATABASE：Cloud Spanner 元数据数据库
  • SPANNER_CHANGE_STREAM：Cloud Spanner 变更数据流
  • GCS_OUTPUT_DIRECTORY：变更数据流输出的文件位置

如需使用 REST API 来运行模板，请发送 HTTP POST 请求。如需详细了解 API 及其授权范围，请参阅 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",
   }
}

替换以下内容：

  • PROJECT_ID：您要在其中运行 Dataflow 作业的 Google Cloud 项目 ID
  • JOB_NAME：您选择的唯一性作业名称
  • VERSION：您要使用的模板的版本

    您可使用以下值：

  • LOCATION：要在其中部署 Dataflow 作业的区域，例如 us-central1
  • SPANNER_INSTANCE_ID：Cloud Spanner 实例 ID
  • SPANNER_DATABASE：Cloud Spanner 数据库
  • SPANNER_METADATA_INSTANCE_ID：Cloud Spanner 元数据实例 ID
  • SPANNER_METADATA_DATABASE：Cloud Spanner 元数据数据库
  • SPANNER_CHANGE_STREAM：Cloud Spanner 变更数据流
  • GCS_OUTPUT_DIRECTORY：变更数据流输出的文件位置
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();
  }
}

后续步骤