Sourcedb to Spanner 範本

Beta 版

這項功能適用《服務專屬條款》中「一般服務條款」一節的《正式發布前產品條款》。正式發布前功能是依「原樣」提供，支援服務可能受限。 詳情請參閱推出階段說明。

「SourceDB 到 Spanner」範本是一個批次管道，可將資料從關聯資料庫複製到現有的 Spanner 資料庫，並且使用 JDBC 來連結關聯式資料庫。您可以使用這個範本，將任何具備可用 JDBC 驅動程式的關聯式資料庫資料複製到 Spanner。這項功能僅支援部分 MySQL 類型

為了增加多一層安全防護，您也可以傳遞 Cloud KMS 金鑰，以及使用 Cloud KMS 金鑰加密的 Base64 編碼使用者名稱、密碼和連線字串參數。如要進一步瞭解如何加密使用者名稱、密碼和連線字串參數，請參閱 Cloud KMS API 加密端點。

注意：如果您打算使用「Cloud Storage Avro 檔案至 Spanner 資料驗證」範本驗證這項遷移作業，就必須填入 gcsOutputDirectory 參數，產生必要的 Avro 檔案。

管道相關規定

• 關聯式資料庫的 JDBC 驅動程式必須可供使用。
• 執行管道前，Spanner 資料表必須已存在。
• Spanner 資料表必須具備相容的結構定義。
• 關聯式資料庫必須可從 Dataflow 執行的子網路存取。

範本參數

必要參數

• sourceConfigURL：來源連線設定檔的網址。檔案格式取決於來源類型。如果是 Astra，則會指向 Astra 連線設定檔 (範例)。如果是 JDBC，則會指向 JDBC 分片設定檔 (範例)。如果是 Cassandra，則會指向 Cassandra 驅動程式設定檔 (範例)。這是必要參數。例如 gs://your-bucket/source-config.json。預設為空白。
• instanceId：目的地 Cloud Spanner 執行個體。
• databaseId：目的地 Cloud Spanner 資料庫。
• projectId：這是 Cloud Spanner 專案的名稱。
• outputDirectory：這個目錄用於傾印遷移作業中失敗/略過/篩選的記錄。

選用參數

• sourceDbDialect：可能的值為 CASSANDRA、MYSQL、POSTGRESQL、ORACLE 和 SQLSERVER。預設值為：MYSQL。
• jdbcDriverJars：以半形逗號分隔的驅動程式 JAR 檔案清單。例如，gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar。預設為空白。
• jdbcDriverClassName：JDBC 驅動程式類別名稱。例如，com.mysql.jdbc.Driver。預設值為：com.mysql.jdbc.Driver。
• 資料表：要從來源遷移的資料表。預設為空白。
• numPartitions：分區數量。這項值會與上下限一起形成分區步幅，用於產生 WHERE 子句運算式，平均分割分區資料欄。如果輸入的值小於 1，系統會將數字設為 1。預設為：0。
• fetchSize：從 JDBC 來源讀取資料時，每頁要擷取的資料列數。如未設定，系統會根據工作站機型和預估的資料列大小自動推斷，如果無法推斷 (例如未指定工作站機型)，則會改為 50,000 列。如果來源方言是 MySQL，請參閱下方注意事項。這最終會轉換為 Jdbc 層的 Statement.setFetchSize 呼叫。只有在預設值擲回記憶體錯誤時，才應使用這個屬性。MySQL 來源注意事項：除非 useCursorFetch=true 也是連線屬性的一部分，否則 MySQL 連接器會忽略 FetchSize。如果是 MySQL 方言，除非 fetchSize 參數明確設為 0，否則管道會預設在連線屬性中加入 useCursorFetch=true。
• spannerHost：要在範本中呼叫的 Cloud Spanner 端點。例如，https://batch-spanner.googleapis.com。預設值為：https://batch-spanner.googleapis.com。
• maxConnections：在每個工作站上設定 JDBC 連線集區，並指定連線數量上限。如要設為無限制，請使用負數。例如，-1。預設為：0。
• sessionFilePath：Cloud Storage 中的工作階段檔案路徑，內含 Spanner 遷移工具的對應資訊。預設為空白。
• transformationJarPath：Cloud Storage 中的自訂 JAR 位置，內含用於處理記錄的自訂轉換邏輯。預設為空白。
• transformationClassName：具有自訂轉換邏輯的完整類別名稱。如果指定 transformationJarPath，這個欄位就必須填寫。預設為空白。
• transformationCustomParameters：字串，內含要傳遞至自訂轉換類別的任何自訂參數。預設為空白。
• insertOnlyModeForSpannerMutations：根據預設，管道會使用 Upsert 將資料列寫入 Spanner。也就是說，現有資料列會遭到覆寫。如果啟用 InsertOnly 模式，系統會使用插入作業，而非 upsert 作業，且不會覆寫現有資料列。
• batchSizeForSpannerMutations：Spanner Mutations 的批次大小 (以位元組為單位)。如果設定的值小於 0，系統會使用 Apache Beam 的 SpannerIO 預設值 (1 MB)。如要停用批次處理變動，請將此值設為 0 或 10。
• spannerPriority：Cloud Spanner 呼叫的要求優先順序。值必須是 [HIGH、MEDIUM、LOW] 其中之一。預設為 HIGH。
• tableOverrides：這是從來源到 Spanner 的資料表名稱覆寫。格式如下：[{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}]這個範例顯示將 Singers 資料表對應至 Vocalists，並將 Albums 資料表對應至 Records。例如，[{Singers, Vocalists}, {Albums, Records}]。預設為空白。
• columnOverrides：這是從來源到 Spanner 的資料欄名稱覆寫。格式如下：[{SourceTableName1.SourceColumnName1, SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1, SourceTableName2.SpannerColumnName1}]請注意，來源和 Spanner 配對中的 SourceTableName 應保持不變。如要覆寫資料表名稱，請使用 tableOverrides。以下範例分別在 Singers 和 Albums 資料表中，將 SingerName 對應至 TalentName，並將 AlbumName 對應至 RecordName。例如，[{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]。預設為空白。
• schemaOverridesFilePath：這個檔案會指定從來源到 Spanner 的資料表和資料欄名稱覆寫。預設為空白。
• uniformizationStageCountHint：標準化階段數提示。目前僅適用於以 JDBC 為基礎的來源，例如 MySQL 或 PostgreSQL。如要停用統一化功能，請保留 0 或預設值。如要設定 log(numPartition) 個階段，請設為 -1。如果來源主鍵空間平均分配 (例如有稀疏洞的自動遞增鍵)，建議保持停用。如果鍵空間不一致，您可能會在資料流執行作業中遇到落後的 VM。在這種情況下，您可以將其設為 -1 來啟用統一化。手動將值設為 0 或 -1 以外的值，有助於微調統一化階段新增的額外負荷，以及因工作分配更完善而帶來的效能提升。
• failureInjectionParameter：失敗注入參數。僅供測試。預設為空白。
• maxCommitDelay：在 Spanner 中，為最佳化寫入處理量而設定的交易延遲時間上限。請參閱 https://cloud.google.com/spanner/docs/throughput-optimized-writes.Set -1，讓 Spanner 選擇預設值。設為正值，即可覆寫，以取得最適合的處理量與延遲時間之間的取捨。預設值為 -1。
• gcsOutputDirectory：這個目錄用於寫入從來源讀取的記錄 AVRO 檔案。例如，gs://your-bucket/your-path。預設為空白。
• disabledAlgorithms：以半形逗號分隔要停用的演算法。如果將這個值設為 none，系統就不會停用任何演算法。請謹慎使用這個參數，因為預設停用的演算法可能存在安全漏洞或效能問題。例如：SSLv3, RC4。
• extraFilesToStage：以半形逗號分隔的 Cloud Storage 路徑或 Secret Manager 密鑰，用於在工作站中暫存檔案。這些檔案會儲存在每個工作人員的 /extra_files 目錄中。例如：gs://<BUCKET_NAME>/file.txt,projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<VERSION_ID>。

執行範本

控制台

1. 前往 Dataflow 的「Create job from template」(依據範本建立工作) 頁面。
前往「依範本建立工作」
2. 在「Job name」(工作名稱) 欄位中，輸入專屬工作名稱。
3. 選用：如要使用區域端點，請從下拉式選單中選取值。預設區域為 us-central1。

   如需可執行 Dataflow 工作的區域清單，請參閱「Dataflow 位置」。

4. 從「Dataflow template」(Dataflow 範本) 下拉式選單中，選取「Sourcedb to Spanner」(從 SourceDB 遷移至 Spanner) 範本。
5. 在提供的參數欄位中輸入參數值。
6. 按一下「Run Job」(執行工作)。

gcloud CLI

注意：如要使用 the Google Cloud CLI 執行 flex範本，您必須安裝 Google Cloud CLI 284.0.0 以上版本。

在殼層或終端機中執行範本：

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates/VERSION/flex/Sourcedb_to_Spanner_Flex \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --parameters \
       sourceConfigURL=SOURCE_CONFIG_URL,\
       instanceId=INSTANCE_ID,\
       databaseId=DATABASE_ID,\
       projectId=PROJECT_ID,\
       outputDirectory=OUTPUT_DIRECTORY,\

更改下列內容：

• JOB_NAME：您選擇的不重複工作名稱
• VERSION：您要使用的範本版本

  您可以使用下列值：

  • latest，使用範本的最新版本，該版本位於 bucket 中非依日期命名的上層資料夾：gs://dataflow-templates/latest/
  • 版本名稱，例如 2023-09-12-00_RC00，可使用特定版本的範本，該範本會以巢狀結構存放在 bucket 中相應的依日期命名上層資料夾中：gs://dataflow-templates/
  注意：範本的最新版本可能會隨著破壞性變更而有所更新。您的正式環境使用的範本應該來自最近依日期命名的上層資料夾，以免這些破壞性變更影響正式環境流程。
• REGION_NAME：您要部署 Dataflow 工作的區域，例如 us-central1
• SOURCE_CONFIG_URL：用來連線至來源資料庫主機的網址。可以是 1. JDBC 連線網址 - 必須包含主機、通訊埠和來源資料庫名稱，且可選擇性包含 autoReconnect、maxReconnects 等屬性。格式：`jdbc:mysql://{host}:{port}/{dbName}?{parameters}`2. 分片設定路徑
• INSTANCE_ID：Cloud Spanner 執行個體 ID。
• DATABASE_ID：Cloud Spanner 資料庫 ID。
• PROJECT_ID：Cloud Spanner 專案 ID。
• OUTPUT_DIRECTORY：失敗/略過/篩選事件的輸出目錄

API

如要使用 REST API 執行範本，請傳送 HTTP POST 要求。如要進一步瞭解 API 和授權範圍，請參閱 projects.templates.launch。

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launchParameter": {
     "jobName": "JOB_NAME",
     "parameters": {
       "sourceConfigURL": "SOURCE_CONFIG_URL",
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "projectId": "PROJECT_ID",
       "outputDirectory": "OUTPUT_DIRECTORY",
     },
     "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/Sourcedb_to_Spanner_Flex",
     "environment": { "maxWorkers": "10" }
  }
}

更改下列內容：

• PROJECT_ID：您要執行 Dataflow 工作的 Google Cloud 專案 ID
• JOB_NAME：您選擇的不重複工作名稱
• VERSION：您要使用的範本版本

  您可以使用下列值：

  • latest，使用範本的最新版本，該版本位於 bucket 中非依日期命名的上層資料夾：gs://dataflow-templates/latest/
  • 版本名稱，例如 2023-09-12-00_RC00，可使用特定版本的範本，該範本會以巢狀結構存放在 bucket 中相應的依日期命名上層資料夾中：gs://dataflow-templates/
  注意：範本的最新版本可能會隨著破壞性變更而有所更新。您的正式環境使用的範本應該來自最近依日期命名的上層資料夾，以免這些破壞性變更影響正式環境流程。
• LOCATION：您要部署 Dataflow 工作的區域，例如 us-central1
• SOURCE_CONFIG_URL：用來連線至來源資料庫主機的網址。可以是 1. JDBC 連線網址 - 必須包含主機、通訊埠和來源資料庫名稱，且可選擇性包含 autoReconnect、maxReconnects 等屬性。格式：`jdbc:mysql://{host}:{port}/{dbName}?{parameters}`2. 分片設定路徑
• INSTANCE_ID：Cloud Spanner 執行個體 ID。
• DATABASE_ID：Cloud Spanner 資料庫 ID。
• PROJECT_ID：Cloud Spanner 專案 ID。
• OUTPUT_DIRECTORY：失敗/略過/篩選事件的輸出目錄

範本原始碼

Java

/*
 * Copyright (C) 2024 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.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.CommonTemplateJvmInitializer;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.options.SourceDbToSpannerOptions;
import com.google.cloud.teleport.v2.source.ISrcToSpSourceConnector;
import com.google.cloud.teleport.v2.source.SourceConnectorFactory;
import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard;
import com.google.cloud.teleport.v2.spanner.migrations.source.config.JdbcShardConfig;
import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConnectionConfig;
import com.google.cloud.teleport.v2.spanner.migrations.utils.DataflowWorkerMachineTypeUtils;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.util.Optional;
import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions;
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.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.commons.lang3.StringUtils;

/**
 * A template that copies data from a relational database using JDBC to an existing Spanner
 * database.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/sourcedb-to-spanner/README_Sourcedb_to_Spanner_Flex.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Sourcedb_to_Spanner_Flex",
    category = TemplateCategory.BATCH,
    displayName = "Sourcedb to Spanner",
    description = {
      "The SourceDB to Spanner template is a batch pipeline that copies data from a relational"
          + " database into an existing Spanner database. This pipeline uses JDBC to connect to"
          + " the relational database. You can use this template to copy data from any relational"
          + " database with available JDBC drivers into Spanner. This currently only supports a limited set of types of MySQL",
    },
    optionsClass = SourceDbToSpannerOptions.class,
    flexContainerName = "source-db-to-spanner",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/sourcedb-to-spanner",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The JDBC drivers for the relational database must be available.",
      "The Spanner tables must exist before pipeline execution.",
      "The Spanner tables must have a compatible schema.",
      "The relational database must be accessible from the subnet where Dataflow runs."
    })
public class SourceDbToSpanner {

  /**
   * Main entry point for executing the pipeline. This will run the pipeline asynchronously. If
   * blocking execution is required, use the {@link SourceDbToSpanner#run} method to start the
   * pipeline and invoke {@code result.waitUntilFinish()} on the {@link PipelineResult}.
   *
   * @param args The command-line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    SourceDbToSpannerOptions options = getSourceDbToSpannerOptions(args);
    run(options);
  }

  @VisibleForTesting
  protected static SourceDbToSpannerOptions getSourceDbToSpannerOptions(String[] args) {
    // Parse the user options passed from the command-line
    SourceDbToSpannerOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(SourceDbToSpannerOptions.class);
    // Stage SSL certificates to extraFiles if required as per the pipeline options.
    // Ref https://cloud.google.com/dataflow/docs/guides/templates/ssl-certificates
    new CommonTemplateJvmInitializer().beforeProcessing(options);
    return options;
  }

  /**
   * Validates the provided pipeline options. TODO: move this to source connector.
   *
   * @param options The execution parameters to the pipeline.
   * @param sourceConnectionConfig Parsed source connection config.
   * @throws IllegalArgumentException if the provided options are invalid for the pipeline.
   */
  @VisibleForTesting
  static void validateOptions(
      SourceDbToSpannerOptions options, SourceConnectionConfig sourceConnectionConfig) {
    if (SourceDbToSpannerOptions.PG_SOURCE_DIALECT.equals(options.getSourceDbDialect())) {
      Preconditions.checkArgument(
          (sourceConnectionConfig instanceof JdbcShardConfig),
          "Postgresql dialect should have JDBC source config.");
      for (Shard shard : ((JdbcShardConfig) sourceConnectionConfig).getShardConfigs()) {
        if (StringUtils.isNotBlank(shard.getNamespace())
            && !shard.getNamespace().equals("public")) {
          throw new IllegalArgumentException(
              "Non-public namespaces are currently unsupported for PostgreSQL migrations.");
        }
      }
    }
  }

  /**
   * Create the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  @VisibleForTesting
  static PipelineResult run(SourceDbToSpannerOptions options) {
    Pipeline pipeline = Pipeline.create(options);
    String workerMachineType =
        pipeline.getOptions().as(DataflowPipelineWorkerPoolOptions.class).getWorkerMachineType();
    Optional<Integer> resourceHintsMinCpus =
        DataflowWorkerMachineTypeUtils.getMinCpuResourceHint(pipeline.getOptions());
    if (!pipeline.getOptions().getRunner().getSimpleName().equals("DirectRunner")) {
      DataflowWorkerMachineTypeUtils.validateMachineSpecs(
          workerMachineType, 4, resourceHintsMinCpus);
    }

    SpannerConfig spannerConfig = createSpannerConfig(options);
    SourceConnectionConfig sourceConnectionConfig =
        PipelineController.getSourceConnectionConfig(
            options.getSourceDbDialect(), options.getSourceConfigURL());

    validateOptions(options, sourceConnectionConfig);

    // Decide type and source of migration
    ISrcToSpSourceConnector connector = SourceConnectorFactory.getSourceConnectorByDialect(options);
    return connector.executeMigration(options, sourceConnectionConfig, pipeline, spannerConfig);
  }

  @VisibleForTesting
  static SpannerConfig createSpannerConfig(SourceDbToSpannerOptions options) {
    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withProjectId(ValueProvider.StaticValueProvider.of(options.getProjectId()))
            .withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()))
            .withInstanceId(ValueProvider.StaticValueProvider.of(options.getInstanceId()))
            .withDatabaseId(ValueProvider.StaticValueProvider.of(options.getDatabaseId()))
            .withRpcPriority(ValueProvider.StaticValueProvider.of(options.getSpannerPriority()));
    if (options.getMaxCommitDelay() >= 0) {
      spannerConfig = spannerConfig.withMaxCommitDelay(options.getMaxCommitDelay());
    }
    return spannerConfig;
  }
}