Pub/Sub Subscription to BigQuery 模板

Pub/Sub Subscription to BigQuery 模板是一种流处理流水线，可从 Pub/Sub 订阅读取 JSON 格式的消息并将其写入 BigQuery 表格中。您可以使用该模板作为将 Pub/Sub 数据移动到 BigQuery 的快速解决方案。此模板可从 Pub/Sub 中读取 JSON 格式的消息并将其转换为 BigQuery 元素。

流水线要求

  • Pub/Sub 消息的 data 字段必须使用 JSON 格式（如此 JSON 指南中所述）。例如，您可以将 data 字段中值为 {"k1":"v1", "k2":"v2"} 格式的消息插入具有名为 k1 和 k2 的列且数据类型为字符串的 BigQuery 表中。
  • 在运行该流水线之前，输出表必须已存在。表架构必须与输入 JSON 对象相匹配。

模板参数

参数 说明
inputSubscription 要读取的 Pub/Sub 输入订阅，格式为 projects/<project>/subscriptions/<subscription>。
outputTableSpec BigQuery 输出表位置，格式为 <my-project>:<my-dataset>.<my-table>
outputDeadletterTable 未能到达输出表的消息的 BigQuery 表，格式为 <my-project>:<my-dataset>.<my-table>。如果该表不存在，则系统会在流水线执行期间创建它。如果未指定此参数，则系统会改用 OUTPUT_TABLE_SPEC_error_records。
javascriptTextTransformGcsPath （可选）.js 文件的 Cloud Storage URI，用于定义您要使用的 JavaScript 用户定义的函数 (UDF)。例如 gs://my-bucket/my-udfs/my_file.js。
javascriptTextTransformFunctionName （可选） 您要使用的 JavaScript 用户定义的函数 (UDF) 的名称。 例如，如果您的 JavaScript 函数代码为 myTransform(inJson) { /*...do stuff...*/ }，则函数名称为 myTransform。如需查看 JavaScript UDF 示例，请参阅 UDF 示例。

用户定义的函数

（可选）您可以通过编写用户定义的函数 (UDF) 来扩展此模板。该模板会为每个输入元素调用 UDF。元素载荷会序列化为 JSON 字符串。如需了解详情，请参阅为 Dataflow 模板创建用户定义的函数。

函数规范

UDF 具有以下规范：

  • 输入：Pub/Sub 消息数据字段，序列化为 JSON 字符串。
  • 输出：与 BigQuery 目标表的架构匹配的 JSON 字符串。

    • 运行模板

    控制台

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

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

    4. 从 Dataflow 模板下拉菜单中，选择 the Pub/Sub Subscription to BigQuery template。
    5. 在提供的参数字段中，输入您的参数值。
    6. 可选：如需从“正好一次”处理切换到“至少一次”流处理模式，请选择至少一次。
    7. 点击运行作业。

    gcloud

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

    gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-REGION_NAME/VERSION/PubSub_Subscription_to_BigQuery \
    --region REGION_NAME \
    --staging-location STAGING_LOCATION \
    --parameters \
inputSubscription=projects/PROJECT_ID/subscriptions/SUBSCRIPTION_NAME,\
outputTableSpec=PROJECT_ID:DATASET.TABLE_NAME,\
outputDeadletterTable=PROJECT_ID:DATASET.TABLE_NAME

    请替换以下内容：

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

      您可使用以下值：

    • STAGING_LOCATION：暂存本地文件的位置（例如 gs://your-bucket/staging）
    • SUBSCRIPTION_NAME：您的 Pub/Sub 订阅名称
    • DATASET：您的 BigQuery 数据集
    • TABLE_NAME：您的 BigQuery 表名称

    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-LOCATION/VERSION/PubSub_Subscription_to_BigQuery
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputSubscription": "projects/PROJECT_ID/subscriptions/SUBSCRIPTION_NAME",
       "outputTableSpec": "PROJECT_ID:DATASET.TABLE_NAME"
   },
   "environment": {
       "ipConfiguration": "WORKER_IP_UNSPECIFIED",
       "additionalExperiments": []
   },
}

    请替换以下内容：

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

      您可使用以下值：

    • STAGING_LOCATION：暂存本地文件的位置（例如 gs://your-bucket/staging）
    • SUBSCRIPTION_NAME：您的 Pub/Sub 订阅名称
    • DATASET：您的 BigQuery 数据集
    • TABLE_NAME：您的 BigQuery 表名称

    后续步骤