Cloud Storage Text 到 Datastore 範本

Beta 版

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

Cloud Storage Text 到 Datastore 範本已淘汰，並於 2023 年第 3 季移除。建議改用「Cloud Storage Text to Firestore」範本。

這個範本是批次管道，可從 Cloud Storage 讀取文字檔案，並將 JSON 編碼的實體寫入 Datastore。輸入文字檔案中的每一行都必須採用指定 JSON 格式。

管道相關規定

• 目的地專案中的 Datastore 必須已啟用。

範本參數

必要參數

• textReadPattern：用來指定文字資料檔案位置的 Cloud Storage 路徑模式，例如：gs://mybucket/somepath/*.json。
• datastoreWriteProjectId：要寫入 Datastore 實體的 Google Cloud 雲端專案 ID。
• errorWritePath：處理期間寫入失敗的錯誤記錄輸出檔案，例如：gs://your-bucket/errors/。

選用參數

• javascriptTextTransformGcsPath：定義要使用的 JavaScript user-defined function (UDF) 的 .js 檔案 Cloud Storage URI。例如：gs://my-bucket/my-udfs/my_file.js。
• javascriptTextTransformFunctionName：要使用的 JavaScript user-defined function (UDF) 名稱。例如，如果您的 JavaScript 函式程式碼是 myTransform(inJson) { /*...do stuff...*/ }，則函式名稱就是 myTransform。如需 JavaScript UDF 範例，請參閱 UDF 範例 (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples)。
• datastoreHintNumWorkers：Datastore 升速節流步驟中預期 worker 數量的提示。預設值為 500。

執行範本

控制台

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

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

4. 從「Dataflow template」(Dataflow 範本) 下拉式選單中，選取「Text Files on Cloud Storage to Datastore」(Cloud Storage 文字檔案至 Datastore) 範本。
5. 在提供的參數欄位中輸入參數值。
6. 按一下「Run Job」(執行工作)。

gcloud

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

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

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/GCS_Text_to_Datastore \
    --region REGION_NAME \
    --parameters \
textReadPattern=PATH_TO_INPUT_TEXT_FILES,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
datastoreWriteProjectId=PROJECT_ID,\
errorWritePath=ERROR_FILE_WRITE_PATH

更改下列內容：

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

  您可以使用下列值：

  • latest，使用範本的最新版本，該版本位於 bucket 中非依日期命名的上層資料夾：gs://dataflow-templates/latest/
  • 版本名稱，例如 2023-09-12-00_RC00，可使用特定版本的範本，該範本會以巢狀結構存放在 bucket 中相應的依日期命名上層資料夾中：gs://dataflow-templates/
  注意：範本的最新版本可能會隨著破壞性變更而有所更新。您的正式環境使用的範本應該來自最近依日期命名的上層資料夾，以免這些破壞性變更影響正式環境流程。
• REGION_NAME：您要部署 Dataflow 工作的區域，例如 us-central1
• PATH_TO_INPUT_TEXT_FILES：Cloud Storage 中的輸入檔案模式
• JAVASCRIPT_FUNCTION： 名稱

  例如，如果您的 JavaScript 函式程式碼是 myTransform(inJson) { /*...do stuff...*/ }，則函式名稱就是 myTransform。如需 JavaScript UDF 範例，請參閱「UDF 範例」。

• PATH_TO_JAVASCRIPT_UDF_FILE： 定義要使用的 JavaScript user-defined function (UDF) 的 .js 檔案 Cloud Storage URI，例如 gs://my-bucket/my-udfs/my_file.js
• ERROR_FILE_WRITE_PATH：Cloud Storage 中錯誤檔案的所需路徑

API

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/GCS_Text_to_Datastore
{
   "jobName": "JOB_NAME",
   "parameters": {
       "textReadPattern": "PATH_TO_INPUT_TEXT_FILES",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION",
       "datastoreWriteProjectId": "PROJECT_ID",
       "errorWritePath": "ERROR_FILE_WRITE_PATH"
   },
   "environment": { "zone": "us-central1-f" }
}

更改下列內容：

• 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
• PATH_TO_INPUT_TEXT_FILES：Cloud Storage 中的輸入檔案模式
• JAVASCRIPT_FUNCTION： 名稱

  例如，如果您的 JavaScript 函式程式碼是 myTransform(inJson) { /*...do stuff...*/ }，則函式名稱就是 myTransform。如需 JavaScript UDF 範例，請參閱「UDF 範例」。

• PATH_TO_JAVASCRIPT_UDF_FILE： 定義要使用的 JavaScript user-defined function (UDF) 的 .js 檔案 Cloud Storage URI，例如 gs://my-bucket/my-udfs/my_file.js
• ERROR_FILE_WRITE_PATH：Cloud Storage 中錯誤檔案的所需路徑

範本原始碼

Java

/*
 * Copyright (C) 2018 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.templates;

import com.google.cloud.teleport.metadata.MultiTemplate;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.templates.TextToDatastore.TextToDatastoreOptions;
import com.google.cloud.teleport.templates.common.DatastoreConverters.DatastoreWriteOptions;
import com.google.cloud.teleport.templates.common.DatastoreConverters.WriteJsonEntities;
import com.google.cloud.teleport.templates.common.ErrorConverters.ErrorWriteOptions;
import com.google.cloud.teleport.templates.common.ErrorConverters.LogErrors;
import com.google.cloud.teleport.templates.common.FirestoreNestedValueProvider;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import com.google.cloud.teleport.templates.common.TextConverters.FilesystemReadOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.values.TupleTag;

/**
 * Dataflow template which reads from a Text Source and writes JSON encoded Entities into Datastore.
 * The JSON is expected to be in the format of: <a
 * href="https://cloud.google.com/datastore/docs/reference/rest/v1/Entity">Entity</a>
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v1/README_GCS_Text_to_Datastore.md">README
 * Datastore</a> or <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v1/README_GCS_Text_to_Firestore.md">README
 * Firestore</a> for instructions on how to use or modify this template.
 */
@MultiTemplate({
  @Template(
      name = "GCS_Text_to_Datastore",
      category = TemplateCategory.LEGACY,
      displayName = "Text Files on Cloud Storage to Datastore [Deprecated]",
      description =
          "The Cloud Storage Text to Datastore template is a batch pipeline that reads from text files stored in "
              + "Cloud Storage and writes JSON encoded Entities to Datastore. "
              + "Each line in the input text files must be in the <a href=\"https://cloud.google.com/datastore/docs/reference/rest/v1/Entity\">specified JSON format</a>.",
      optionsClass = TextToDatastoreOptions.class,
      skipOptions = {
        "firestoreWriteProjectId",
        "firestoreWriteEntityKind",
        "firestoreWriteNamespace",
        "firestoreHintNumWorkers",
        "javascriptTextTransformReloadIntervalMinutes",
        // This template doesn't use neither firestoreWriteEntityKind/Namespace
        // nor datastoreWriteEntityKind/Namespace pipeline options.
        "datastoreWriteEntityKind",
        "datastoreWriteNamespace"
      },
      documentation =
          "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-storage-to-datastore",
      contactInformation = "https://cloud.google.com/support",
      preview = true,
      requirements = {"Datastore must be enabled in the destination project."}),
  @Template(
      name = "GCS_Text_to_Firestore",
      category = TemplateCategory.BATCH,
      displayName = "Text Files on Cloud Storage to Firestore (Datastore mode)",
      description =
          "The Cloud Storage Text to Firestore template is a batch pipeline that reads from text files stored in "
              + "Cloud Storage and writes JSON encoded Entities to Firestore. "
              + "Each line in the input text files must be in the <a href=\"https://cloud.google.com/datastore/docs/reference/rest/v1/Entity\">specified JSON format</a>.",
      optionsClass = TextToDatastoreOptions.class,
      skipOptions = {
        "datastoreWriteProjectId",
        "datastoreWriteEntityKind",
        "datastoreWriteNamespace",
        "datastoreHintNumWorkers",
        "javascriptTextTransformReloadIntervalMinutes",
        // This template doesn't use neither firestoreWriteEntityKind/Namespace
        // nor datastoreWriteEntityKind/Namespace pipeline options.
        "firestoreWriteEntityKind",
        "firestoreWriteNamespace"
      },
      documentation =
          "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-storage-to-firestore",
      contactInformation = "https://cloud.google.com/support",
      requirements = {"Firestore must be enabled in the destination project."})
})
public class TextToDatastore {

  public static <T> ValueProvider<T> selectProvidedInput(
      ValueProvider<T> datastoreInput, ValueProvider<T> firestoreInput) {
    return new FirestoreNestedValueProvider(datastoreInput, firestoreInput);
  }

  /** TextToDatastore Pipeline Options. */
  public interface TextToDatastoreOptions
      extends PipelineOptions,
          FilesystemReadOptions,
          JavascriptTextTransformerOptions,
          DatastoreWriteOptions,
          ErrorWriteOptions {}

  /**
   * Runs a pipeline which reads from a Text Source, passes the Text to a Javascript UDF, writes the
   * JSON encoded Entities to a TextIO sink.
   *
   * <p>If your Text Source does not contain JSON encoded Entities, then you'll need to supply a
   * Javascript UDF which transforms your data to be JSON encoded Entities.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    TextToDatastoreOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(TextToDatastoreOptions.class);

    TupleTag<String> errorTag = new TupleTag<String>("errors") {};

    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply(TextIO.read().from(options.getTextReadPattern()))
        .apply(
            TransformTextViaJavascript.newBuilder()
                .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                .setFunctionName(options.getJavascriptTextTransformFunctionName())
                .build())
        .apply(
            WriteJsonEntities.newBuilder()
                .setProjectId(
                    selectProvidedInput(
                        options.getDatastoreWriteProjectId(), options.getFirestoreWriteProjectId()))
                .setHintNumWorkers(
                    selectProvidedInput(
                        options.getDatastoreHintNumWorkers(), options.getFirestoreHintNumWorkers()))
                .setErrorTag(errorTag)
                .build())
        .apply(
            LogErrors.newBuilder()
                .setErrorWritePath(options.getErrorWritePath())
                .setErrorTag(errorTag)
                .build());

    pipeline.run();
  }
}

後續步驟

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