Spanner 到 BigQuery 範本

Beta 版

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

「Spanner 到 BigQuery」範本為批次管道，可從 Spanner 資料表讀取資料，並將資料寫入 BigQuery。

管道相關規定

• 執行管道之前，來源 Spanner 資料表必須已存在。
• 執行管道之前，BigQuery 資料集必須存在。
• 說明您 BigQuery 結構定義的 JSON 檔案。

  檔案必須包含名為「fields」的頂層 JSON 陣列。 fields 陣列的內容必須採用以下格式：
  {"name": "COLUMN_NAME", "type": "DATA_TYPE"}。

  下列 JSON 說明 BigQuery 結構定義範例：

  {
    "fields": [
      {
        "name": "location",
        "type": "STRING"
      },
      {
        "name": "name",
        "type": "STRING"
      },
      {
        "name": "age",
        "type": "STRING"
      },
      {
        "name": "color",
        "type": "STRING"
      },
      {
        "name": "coffee",
        "type": "STRING"
      }
    ]
  }

  「將 Spanner 資料批次匯入 BigQuery」範本不支援將資料匯入目標 BigQuery 資料表的 STRUCT (記錄) 欄位。

範本參數

必要參數

• spannerInstanceId：要讀取的 Spanner 資料庫執行個體 ID。
• spannerDatabaseId：要匯出的 Spanner 資料庫 ID。
• outputTableSpec：用於寫入輸出內容的 BigQuery 輸出資料表位置。例如：<PROJECT_ID>:<DATASET_NAME>.<TABLE_NAME>。視指定的 createDisposition 而定，系統可能會使用使用者提供的 Avro 結構定義，自動建立輸出資料表。

選用參數

• spannerProjectId：Spanner 資料庫所在的專案 ID。這個參數的預設值是執行 Dataflow 管道的專案。
• spannerTableId：要匯出的 Spanner 資料庫資料表名稱。如果已設定 sqlQuery，系統會忽略這項屬性。
• spannerRpcPriority：Spanner 呼叫的要求優先順序 (https://cloud.google.com/spanner/docs/reference/rest/v1/RequestOptions)。可能的值為 HIGH、MEDIUM 和 LOW。預設值為 HIGH。
• sqlQuery：用於從 Spanner 資料庫讀取資料的 SQL 查詢。如果 spannerTableId 為空白，則為必要欄位。
• bigQuerySchemaPath：BigQuery 結構定義的 JSON 檔案 Cloud Storage 路徑 (gs://)。如果 Create Disposition 不是 CREATE_NEVER，則為必要。例如 gs://your-bucket/your-schema.json。
• writeDisposition：BigQuery WriteDisposition (https://cloud.google.com/bigquery/docs/reference/rest/v2/Job#jobconfigurationload) 值。例如：WRITE_APPEND、WRITE_EMPTY 或 WRITE_TRUNCATE。預設為 WRITE_APPEND。
• createDisposition：BigQuery CreateDisposition (https://cloud.google.com/bigquery/docs/reference/rest/v2/Job#jobconfigurationload)。例如 CREATE_IF_NEEDED 和 CREATE_NEVER。預設為 CREATE_IF_NEEDED。
• useStorageWriteApi：如果為 true，管道會使用 BigQuery Storage Write API (https://cloud.google.com/bigquery/docs/write-api)。預設值為 false。詳情請參閱「使用 Storage Write API」(https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api)。
• useStorageWriteApiAtLeastOnce：使用 Storage Write API 時，指定寫入語意。如要使用至少一次語意 (https://beam.apache.org/documentation/io/built-in/google-bigquery/#at-least-once-semantics)，請將這個參數設為 true。如要使用僅一次語意，請將參數設為 false。只有在 useStorageWriteApi 為 true 時，這項參數才會生效。預設值為 false。

執行範本

控制台

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

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

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

gcloud

注意：如要使用 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/Cloud_Spanner_to_BigQuery_Flex \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --parameters \
       spannerInstanceId=SPANNER_INSTANCE_ID,\
       spannerDatabaseId=SPANNER_DATABASE_ID,\
       spannerTableId=SPANNER_TABLE_ID,\
       sqlQuery=SQL_QUERY,\
       outputTableSpec=OUTPUT_TABLE_SPEC,\

更改下列內容：

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

  您可以使用下列值：

  • latest，使用範本的最新版本，該版本位於 bucket 中非依日期命名的上層資料夾：gs://dataflow-templates/latest/
  • 版本名稱，例如 2023-09-12-00_RC00，可使用特定版本的範本，該範本會以巢狀結構存放在 bucket 中相應的依日期命名上層資料夾中：gs://dataflow-templates/
  注意：範本的最新版本可能會隨著破壞性變更而有所更新。您的正式環境使用的範本應該來自最近依日期命名的上層資料夾，以免這些破壞性變更影響正式環境流程。
• REGION_NAME：您要部署 Dataflow 工作的區域，例如 us-central1
• SPANNER_INSTANCE_ID：Spanner 執行個體 ID
• SPANNER_DATABASE_ID：Spanner 資料庫 ID
• SPANNER_TABLE_ID：Spanner 資料表名稱
• SQL_QUERY：SQL 查詢
• OUTPUT_TABLE_SPEC：BigQuery 資料表位置

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": {
       "spannerInstanceId": "SPANNER_INSTANCE_ID",
       "spannerDatabaseId": "SPANNER_DATABASE_ID",
       "spannerTableId": "SPANNER_TABLE_ID",
       "sqlQuery": "SQL_QUERY",
       "outputTableSpec": "OUTPUT_TABLE_SPEC",
     },
     "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/Cloud_Spanner_to_BigQuery_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
• SPANNER_INSTANCE_ID：Spanner 執行個體 ID
• SPANNER_DATABASE_ID：Spanner 資料庫 ID
• SPANNER_TABLE_ID：Spanner 資料表名稱
• SQL_QUERY：SQL 查詢
• OUTPUT_TABLE_SPEC：BigQuery 資料表位置

範本原始碼

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 static com.google.cloud.teleport.v2.utils.GCSUtils.getGcsFileAsString;
import static org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition.CREATE_NEVER;

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.SpannerToBigQueryOptions;
import com.google.cloud.teleport.v2.transforms.BigQueryConverters;
import com.google.cloud.teleport.v2.transforms.SpannerToBigQueryTransform.StructToJson;
import com.google.cloud.teleport.v2.utils.BigQueryIOUtils;
import com.google.common.base.Strings;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
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;

/** Template to read data from a Spanner table and write into a BigQuery table. */
@Template(
    name = "Cloud_Spanner_to_BigQuery_Flex",
    category = TemplateCategory.BATCH,
    displayName = "Spanner to BigQuery",
    description =
        "The Spanner to BigQuery template is a batch pipeline that reads data from a Spanner table, and writes them to a BigQuery table.",
    optionsClass = SpannerToBigQueryOptions.class,
    flexContainerName = "googlecloud-to-googlecloud",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/spanner-to-bigquery",
    contactInformation = "https://cloud.google.com/support")
public final class SpannerToBigQuery {

  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    PipelineOptionsFactory.register(SpannerToBigQueryOptions.class);
    SpannerToBigQueryOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(SpannerToBigQueryOptions.class);

    BigQueryIOUtils.validateBQStorageApiOptionsBatch(options);

    Pipeline pipeline = Pipeline.create(options);

    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withProjectId(
                options.getSpannerProjectId().isEmpty()
                    ? options.getProject()
                    : options.getSpannerProjectId())
            .withDatabaseId(options.getSpannerDatabaseId())
            .withInstanceId(options.getSpannerInstanceId())
            .withRpcPriority(options.getSpannerRpcPriority());

    SpannerIO.Read read = SpannerIO.read().withSpannerConfig(spannerConfig);

    if (!Strings.isNullOrEmpty(options.getSqlQuery())) {
      read = read.withQuery(options.getSqlQuery());
    } else if (!Strings.isNullOrEmpty(options.getSpannerTableId())) {
      read = read.withTable(options.getSpannerTableId());
    } else {
      throw new IllegalArgumentException("either sqlQuery or spannerTableId required");
    }
    if (Strings.isNullOrEmpty(options.getBigQuerySchemaPath())
        && CreateDisposition.valueOf(options.getCreateDisposition()) != CREATE_NEVER) {
      throw new IllegalArgumentException(
          "bigQuerySchemaPath is required if CreateDisposition is not CREATE_NEVER");
    }
    pipeline
        .apply(read)
        .apply(new StructToJson())
        .apply("Write To BigQuery", writeToBigQuery(options));

    pipeline.run();
  }

  private static Write<String> writeToBigQuery(SpannerToBigQueryOptions options) {
    if (CreateDisposition.valueOf(options.getCreateDisposition()) == CREATE_NEVER) {
      return BigQueryIO.<String>write()
          .to(options.getOutputTableSpec())
          .withWriteDisposition(WriteDisposition.valueOf(options.getWriteDisposition()))
          .withCreateDisposition(CreateDisposition.valueOf(options.getCreateDisposition()))
          .withExtendedErrorInfo()
          .withFormatFunction(BigQueryConverters::convertJsonToTableRow);
    }
    return BigQueryIO.<String>write()
        .to(options.getOutputTableSpec())
        .withWriteDisposition(WriteDisposition.valueOf(options.getWriteDisposition()))
        .withCreateDisposition(CreateDisposition.valueOf(options.getCreateDisposition()))
        .withExtendedErrorInfo()
        .withFormatFunction(BigQueryConverters::convertJsonToTableRow)
        .withJsonSchema(getGcsFileAsString(options.getBigQuerySchemaPath()));
  }
}

後續步驟

• 瞭解 Dataflow 範本。
• 請參閱 Google 提供的範本清單。