Cloud Storage Text to BigQuery (Stream) 模板

Cloud Storage Text to BigQuery 流水线是一种流处理流水线，用于流式传输 Cloud Storage 中存储的文本文件，使用您提供的 JavaScript 用户定义的函数 (UDF) 转换这些文件，然后将结果附加到 BigQuery。

流水线无限期运行，需要通过取消而非排空手动终止，原因是其使用 Watch 转换，该转换是不支持排空的可拆分 DoFn。

流水线要求

  • 创建一个用于描述 BigQuery 中的输出表架构的 JSON 文件。

    确保有一个名为 fields 的顶级 JSON 数组，且该数组的内容遵循 {"name": "COLUMN_NAME", "type": "DATA_TYPE"} 格式。例如： 

    {
  "fields": [
    {
      "name": "name",
      "type": "STRING"
    },
    {
      "name": "age",
      "type": "INTEGER"
    }
  ]
}
  • 使用 UDF 函数（该函数提供转换文本行的逻辑）创建一个 JavaScript (.js) 文件。您的函数必须返回一个 JSON 字符串。

    以下示例将拆分 CSV 文件的每一行，使用相应值创建 JSON 对象，并返回 JSON 字符串：

    function process(inJson) {
  val = inJson.split(",");

  const obj = {
    "name": val[0],
    "age": parseInt(val[1])
  };
  return JSON.stringify(obj);
}

模板参数

必需参数

  • inputFilePattern：Cloud Storage 中待处理的文本的 gs:// 路径。例如 gs://your-bucket/your-file.txt。
  • JSONPath：用于定义 BigQuery 架构的 JSON 文件的 gs:// 路径（存储在 Cloud Storage 中）。例如 gs://your-bucket/your-schema.json。
  • outputTable：用于存储已处理数据的 BigQuery 表的位置。如果您重复使用现有表，该表会被覆盖。例如 <PROJECT_ID>:<DATASET_NAME>.<TABLE_NAME>。
  • javascriptTextTransformGcsPath：.js 文件的 Cloud Storage URI，用于定义您要使用的 JavaScript 用户定义的函数 (UDF)。例如 gs://your-bucket/your-transforms/*.js。
  • javascriptTextTransformFunctionName：您要使用的 JavaScript 用户定义的函数 (UDF) 的名称。例如，如果 JavaScript 函数代码为 myTransform(inJson) { /*...do stuff...*/ }，则函数名称为 myTransform。如需查看 JavaScript UDF 示例，请参阅 UDF 示例 (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples)。例如，transform_udf1。
  • bigQueryLoadingTemporaryDirectory：BigQuery 加载进程的临时目录。例如 gs://your-bucket/your-files/temp-dir。

可选参数

  • outputDeadletterTable：未能到达输出表的消息表。如果表不存在，系统会在流水线执行期间创建该表。如果未指定，则系统会使用 <outputTableSpec>_error_records。例如 <PROJECT_ID>:<DATASET_NAME>.<TABLE_NAME>。
  • useStorageWriteApiAtLeastOnce：此参数仅在启用了 Use BigQuery Storage Write API 时有效。如果启用，则系统会将“至少一次”语义用于 Storage Write API，否则会使用“正好一次”语义。默认值为：false。
  • 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)。
  • numStorageWriteApiStreams：使用 Storage Write API 时，指定写入流的数量。如果 useStorageWriteApi 为 true 且 useStorageWriteApiAtLeastOnce 为 false，则必须设置此参数。默认值为 0。
  • storageWriteApiTriggeringFrequencySec：使用 Storage Write API 时，指定触发频率（以秒为单位）。如果 useStorageWriteApi 为 true 且 useStorageWriteApiAtLeastOnce 为 false，则必须设置此参数。
  • pythonExternalTextTransformGcsPath：包含用户定义的函数的 Python 代码的 Cloud Storage 路径模式。例如 gs://your-bucket/your-function.py。
  • javascriptTextTransformReloadIntervalMinutes：指定重新加载 UDF 的频率（以分钟为单位）。如果值大于 0，则 Dataflow 会定期检查 Cloud Storage 中的 UDF 文件，并在文件修改时重新加载 UDF。此参数可让您在流水线运行时更新 UDF，而无需重启作业。如果值为 0，则停用 UDF 重新加载。默认值为 0。

用户定义的函数

此模板需要使用 UDF 来解析输入文件，如流水线要求中所述。该模板会为每个输入文件中的每一行文本调用 UDF。如需详细了解如何创建 UDF，请参阅为 Dataflow 模板创建用户定义的函数。

函数规范

UDF 具有以下规范：

  • 输入：来自输入文件的一行文本。
  • 输出：与 BigQuery 目标表的架构匹配的 JSON 字符串。

运行模板

控制台

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

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

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

gcloud

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

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location gs://dataflow-templates-REGION_NAME/VERSION/flex/Stream_GCS_Text_to_BigQuery_Flex \
    --region REGION_NAME \
    --staging-location STAGING_LOCATION \
    --parameters \
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
JSONPath=PATH_TO_BIGQUERY_SCHEMA_JSON,\
inputFilePattern=PATH_TO_TEXT_DATA,\
outputTable=BIGQUERY_TABLE,\
outputDeadletterTable=BIGQUERY_UNPROCESSED_TABLE,\
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS

替换以下内容：

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

    您可使用以下值：

  • STAGING_LOCATION：暂存本地文件的位置（例如 gs://your-bucket/staging）
  • JAVASCRIPT_FUNCTION： 您要使用的 JavaScript 用户定义的函数 (UDF) 的名称

    例如，如果您的 JavaScript 函数代码为 myTransform(inJson) { /*...do stuff...*/ }，则函数名称为 myTransform。如需查看 JavaScript UDF 示例，请参阅 UDF 示例。

  • PATH_TO_BIGQUERY_SCHEMA_JSON：包含架构定义的 JSON 文件的 Cloud Storage 路径
  • PATH_TO_JAVASCRIPT_UDF_FILE： .js 文件的 Cloud Storage URI，用于定义您要使用的 JavaScript 用户定义的函数 (UDF)，例如 gs://my-bucket/my-udfs/my_file.js
  • PATH_TO_TEXT_DATA：文本数据集的 Cloud Storage 路径
  • BIGQUERY_TABLE：您的 BigQuery 表名称
  • BIGQUERY_UNPROCESSED_TABLE：未处理消息的 BigQuery 表名称
  • PATH_TO_TEMP_DIR_ON_GCS：临时目录的 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",
      "parameters": {
       "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION",
       "JSONPath": "PATH_TO_BIGQUERY_SCHEMA_JSON",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "inputFilePattern":"PATH_TO_TEXT_DATA",
       "outputTable":"BIGQUERY_TABLE",
       "outputDeadletterTable":"BIGQUERY_UNPROCESSED_TABLE",
       "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Stream_GCS_Text_to_BigQuery_Flex",
   }
}

替换以下内容：

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

    您可使用以下值：

  • STAGING_LOCATION：暂存本地文件的位置（例如 gs://your-bucket/staging）
  • JAVASCRIPT_FUNCTION： 您要使用的 JavaScript 用户定义的函数 (UDF) 的名称

    例如，如果您的 JavaScript 函数代码为 myTransform(inJson) { /*...do stuff...*/ }，则函数名称为 myTransform。如需查看 JavaScript UDF 示例，请参阅 UDF 示例。

  • PATH_TO_BIGQUERY_SCHEMA_JSON：包含架构定义的 JSON 文件的 Cloud Storage 路径
  • PATH_TO_JAVASCRIPT_UDF_FILE： .js 文件的 Cloud Storage URI，用于定义您要使用的 JavaScript 用户定义的函数 (UDF)，例如 gs://my-bucket/my-udfs/my_file.js
  • PATH_TO_TEXT_DATA：文本数据集的 Cloud Storage 路径
  • BIGQUERY_TABLE：您的 BigQuery 表名称
  • BIGQUERY_UNPROCESSED_TABLE：未处理消息的 BigQuery 表名称
  • PATH_TO_TEMP_DIR_ON_GCS：临时目录的 Cloud Storage 路径

后续步骤