Datastream to Spanner 模板

Datastream to Spanner 模板是一种流处理流水线，可从 Cloud Storage 存储桶中读取 Datastream 事件并将其写入 Spanner 数据库。它适用于从 Datastream 来源到 Spanner 的数据迁移。

在执行模板之前，迁移所需的所有表必须存在于目标 Spanner 数据库中。因此，在数据迁移之前，必须完成从源数据库到目标 Spanner 的架构迁移。在迁移之前，数据可能存在表中。此模板不会将 Datastream 架构更改传播到 Spanner 数据库。

只有在所有数据都写入 Spanner 后，才能在迁移结束时保证数据一致性。为了存储写入 Spanner 的每个记录的排序信息，此模板为 Spanner 数据库中的每个表创建了一个额外的表（称为影子表）。这用于确保迁移结束时的一致性。影子表在迁移后不会被删除，可在迁移结束时用于进行验证。

操作期间发生的任何错误（例如架构不匹配、JSON 文件格式错误或执行转换产生的错误）都会记录在错误队列中。错误队列是一个 Cloud Storage 文件夹，它以文本格式存储遇到错误的所有 Datastream 事件以及错误原因。这些错误可能是暂时性的，也可能是永久性的，它们存储在错误队列的相应 Cloud Storage 文件夹中。系统会自动重试暂时性错误，但不会自动重试永久性错误。如果发生永久性错误，您可以选择在模板运行期间更正更改事件，并将它们转移到可重试的存储桶。

流水线要求

  • 处于正在运行或未启动状态的 Datastream 数据流。
  • 要在其中复制 Datastream 事件的 Cloud Storage 存储桶。
  • 包含现有表的 Spanner 数据库。这些表可以为空，也可以包含数据。

模板参数

必需参数

  • inputFilePattern：包含要复制的 Datastream 文件的 Cloud Storage 文件位置。通常，这是数据流的根路径。
  • instanceId：在其中复制更改的 Spanner 实例。
  • databaseId：在其中复制更改的 Spanner 数据库。

可选参数

  • inputFileFormat：Datastream 生成的输出文件的格式。例如 avro,json。默认值为 avro。
  • sessionFilePath：Cloud Storage 中的会话文件路径，其中包含 HarbourBridge 的映射信息。
  • projectId：Spanner 项目 ID。
  • spannerHost：要在模板中调用的 Cloud Spanner 端点。（示例：https://batch-spanner.googleapis.com）。默认值为：https://batch-spanner.googleapis.com。
  • streamName：用于轮询架构信息和来源类型的数据流的名称或模板。
  • gcsPubSubSubscription：Cloud Storage 通知政策中使用的 Pub/Sub 订阅。该名称应采用 projects/ 格式
  • shadowTablePrefix：用于为影子表命名的前缀。默认值：shadow_。
  • shouldCreateShadowTables：此标志指示是否必须在 Cloud Spanner 数据库中创建影子表。默认值为：true。
  • rfcStartDateTime：用于从 Cloud Storage 中提取数据的起始日期时间 (https://tools.ietf.org/html/rfc3339)。默认值为：1970-01-01T00:00:00.00Z。
  • fileReadConcurrency：要读取的并发 DataStream 文件的数量。默认值为：30。
  • deadLetterQueueDirectory：存储错误队列输出时使用的文件路径。默认文件路径为 Dataflow 作业的临时位置下的目录。
  • dlqRetryMinutes：死信队列重试之间的分钟数。默认值为 10。
  • dlqMaxRetryCount：可通过 DLQ 重试临时错误的次数上限。默认值为 500。
  • dataStreamRootUrl：Datastream API 根网址。默认值为：https://datastream.googleapis.com/。
  • datastreamSourceType：这是 Datastream 连接到的源数据库的类型。示例：mysql/oracle。在测试时，如果没有实际运行的 Datastream，则需要进行设置。
  • roundJsonDecimals：如果设置了此标志，则会将 json 列中的小数值四舍五入为可以在不损失精度的情况下存储的数字。默认值为：false。
  • runMode：这是运行模式类型，无论是常规模式还是具有 retryDLQ 的模式。默认值为：regular。
  • transformationContextFilePath：Cloud Storage 中的转换上下文文件路径，用于填充迁移期间执行转换时使用的数据。示例：用于标识从中迁移行的数据库的分片 ID 到数据库名称。
  • directoryWatchDurationInMinutes：流水线应持续轮询 GCS 中某个目录的时长。Datastream 输出文件以目录结构进行排列，该目录结构描述了按分钟分组的事件时间戳。此参数应大致等于在源数据库中发生的事件与 Datastream 写入 GCS 的同一事件之间可能出现的最长延迟时间。第 99.9 百分位 = 10 分钟。默认值为：10。
  • spannerPriority：Cloud Spanner 调用的请求优先级。该值必须为 [HIGH,MEDIUM,LOW] 之一。默认值为：HIGH。
  • dlqGcsPubSubSubscription：在常规模式下运行时，Cloud Storage 通知政策中用于 DLQ 重试目录的 Pub/Sub 订阅。该名称应采用 projects/ 格式
  • transformationJarPath：Cloud Storage 中的自定义 jar 位置，其中包含用于处理正向迁移中的记录的自定义转换逻辑。默认值为空。
  • transformationClassName：包含自定义转换逻辑的完全限定类名称。如果指定了 transformationJarPath，则这是必填字段。默认值为空。
  • transformationCustomParameters：包含要传递给自定义转换类的任何自定义参数的字符串。默认值为空。
  • filteredEventsDirectory：这是用于存储通过自定义转换过滤的事件的文件路径。默认值为 Dataflow 作业的临时位置下的目录。在大多数情况下，默认值就可以了。
  • shardingContextFilePath：Cloud Storage 中的分片上下文文件路径，用于在迁移期间填充分片 ID。其格式为 Map<stream_name, Map<db_name, shard_id>>。

运行模板

控制台

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

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

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

gcloud

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

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Cloud_Datastream_to_Spanner \
    --parameters \
inputFilePattern=GCS_FILE_PATH,\
streamName=STREAM_NAME,\
instanceId=CLOUDSPANNER_INSTANCE,\
databaseId=CLOUDSPANNER_DATABASE,\
deadLetterQueueDirectory=DLQ

替换以下内容：

  • PROJECT_ID：您要在其中运行 Dataflow 作业的 Google Cloud 项目的 ID
  • JOB_NAME：您选择的唯一性作业名称
  • REGION_NAME：要在其中部署 Dataflow 作业的区域，例如 us-central1
  • VERSION：您要使用的模板的版本

    您可使用以下值：

  • GCS_FILE_PATH：用于存储 Datastream 事件的 Cloud Storage 路径。例如 gs://bucket/path/to/data/
  • CLOUDSPANNER_INSTANCE：Spanner 实例。
  • CLOUDSPANNER_DATABASE：Spanner 数据库。
  • DLQ：错误队列目录的 Cloud Storage 路径。

API

如需使用 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",
      "containerSpecGcsPath": "gs://dataflow-templates-REGION_NAME/VERSION/flex/Cloud_Datastream_to_Spanner",
      "parameters": {
          "inputFilePattern": "GCS_FILE_PATH",
          "streamName": "STREAM_NAME"
          "instanceId": "CLOUDSPANNER_INSTANCE"
          "databaseId": "CLOUDSPANNER_DATABASE"
          "deadLetterQueueDirectory": "DLQ"
      }
   }
}

替换以下内容：

  • PROJECT_ID：您要在其中运行 Dataflow 作业的 Google Cloud 项目的 ID
  • JOB_NAME：您选择的唯一性作业名称
  • LOCATION：要在其中部署 Dataflow 作业的区域，例如 us-central1
  • VERSION：您要使用的模板的版本

    您可使用以下值：

  • GCS_FILE_PATH：用于存储 Datastream 事件的 Cloud Storage 路径。例如 gs://bucket/path/to/data/
  • CLOUDSPANNER_INSTANCE：Spanner 实例。
  • CLOUDSPANNER_DATABASE：Spanner 数据库。
  • DLQ：错误队列目录的 Cloud Storage 路径。

后续步骤