从 Pub/Sub 流式传输到 BigQuery

---

本教程使用 Pub/Sub Subscription to BigQuery 模板，通过 Google Cloud 控制台或 Google Cloud CLI 创建并运行 Dataflow 模板作业。本教程介绍了一个流处理流水线示例，该示例从 Pub/Sub 中读取 JSON 编码的消息，使用用户定义的函数 (UDF) 来扩展 Google 提供的流处理模板，使用 Apache Beam SDK 转换消息数据，然后将结果写入 BigQuery 表中。

此工作流的优势在于，您可以使用 UDF 来扩展 Google 提供的流处理模板。如果您需要从 Pub/Sub 拉取数据并将其输出到 BigQuery，但不需要扩展模板，则对于更简单的工作流，请使用 Pub/Sub to BigQuery 订阅功能。

流式分析和数据集成流水线使用 Pub/Sub 提取和分发数据。通过 Pub/Sub，您可以创建事件提供方和使用方的系统，称为发布者和订阅者。发布者将事件异步发送到 Pub/Sub 服务，Pub/Sub 将事件传递给需要响应事件的所有服务。

Dataflow 是一种全代管式服务，用于以流式传输（实时）和批量模式对数据进行转换并丰富数据内容。它提供了一个简化的流水线开发环境，该环境使用 Apache Beam SDK 转换传入的数据，然后输出转换后的数据。

如果您要将消息直接写入 BigQuery，而不是通过配置 Dataflow 来提供数据转换，那么请使用 Pub/Sub 到 BigQuery 订阅功能。

目标

• 创建 Pub/Sub 主题。
• 使用表和架构创建 BigQuery 数据集。
• 使用 Google 提供的流处理模板，通过 Dataflow 将数据从 Pub/Sub 订阅流式传输到 BigQuery。
• 创建用户定义的函数 (UDF) 以扩展 Google 提供的流处理模板。

费用

在本文档中，您将使用 Google Cloud 的以下收费组件：

• Dataflow
• Pub/Sub
• Cloud Storage
• Cloud Scheduler
• BigQuery

您可使用价格计算器根据您的预计使用情况来估算费用。 Google Cloud 新用户可能有资格申请免费试用。

完成本文档中描述的任务后，您可以通过删除所创建的资源来避免继续计费。如需了解详情，请参阅清理。

准备工作

本部分介绍了如何选择项目、启用 API 以及向您的用户账号和工作器服务账号授予适当的角色。

控制台

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

3. Make sure that billing is enabled for your Google Cloud project.

4. Enable the Compute Engine, Dataflow, Cloud Logging, BigQuery, Pub/Sub, Cloud Storage, Resource Manager, and Cloud Scheduler APIs.

   Enable the APIs

5. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

6. Make sure that billing is enabled for your Google Cloud project.

7. Enable the Compute Engine, Dataflow, Cloud Logging, BigQuery, Pub/Sub, Cloud Storage, Resource Manager, and Cloud Scheduler APIs.

   Enable the APIs

8. 如需完成本教程中的步骤，您的用户账号必须具有 Service Account User 角色。Compute Engine 默认服务账号必须具有以下角色：Dataflow Worker、Dataflow Admin、Pub/Sub Editor、Storage Object Admin 和 BigQuery Data Editor。如需在 Google Cloud 控制台中添加所需的角色，请执行以下操作：

   1. 在 Google Cloud 控制台中，转到 IAM 页面。

      转到 IAM
   2. 选择您的项目。
   3. 在用户账号所在的行中，点击 修改主账号，然后点击 添加其他角色。
   4. 在下拉列表中，选择 Service Account User 角色。
   5. 在 Compute Engine 默认服务账号所在的行中，点击 修改主账号，然后点击 添加其他角色。
   6. 在下拉列表中，选择 Dataflow Worker 角色。

   7. 对 Dataflow Admin、Pub/Sub Editor、Storage Object Admin 和 BigQuery Data Editor 角色重复上述步骤，然后点击保存。

      如需详细了解如何授予角色，请参阅使用控制台授予 IAM 角色。

gcloud

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
2. Install the Google Cloud CLI.

3. To initialize the gcloud CLI, run the following command:

   gcloud init

4. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

5. Make sure that billing is enabled for your Google Cloud project.

6. Enable the Compute Engine, Dataflow, Cloud Logging, BigQuery, Pub/Sub, Cloud Storage, Resource Manager, and Cloud Scheduler APIs:

   gcloud services enable compute.googleapis.com dataflow.googleapis.com logging.googleapis.com bigquery.googleapis.com pubsub.googleapis.com storage.googleapis.com cloudresourcemanager.googleapis.com cloudscheduler.googleapis.com

7. If you're using a local shell, then create local authentication credentials for your user account:

   gcloud auth application-default login

   You don't need to do this if you're using Cloud Shell.

8. Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/iam.serviceAccountUser

   gcloud projects add-iam-policy-binding PROJECT_ID --member="USER_IDENTIFIER" --role=ROLE

   • Replace PROJECT_ID with your project ID.

   • Replace USER_IDENTIFIER with the identifier for your user account. For example, user:[e-mail].

   • Replace ROLE with each individual role.
9. Install the Google Cloud CLI.

10. To initialize the gcloud CLI, run the following command:

    gcloud init

11. Create or select a Google Cloud project.

    Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

    • Create a Google Cloud project:

      gcloud projects create PROJECT_ID

      Replace PROJECT_ID with a name for the Google Cloud project you are creating.

    • Select the Google Cloud project that you created:

      gcloud config set project PROJECT_ID

      Replace PROJECT_ID with your Google Cloud project name.

12. Make sure that billing is enabled for your Google Cloud project.

13. Enable the Compute Engine, Dataflow, Cloud Logging, BigQuery, Pub/Sub, Cloud Storage, Resource Manager, and Cloud Scheduler APIs:

    gcloud services enable compute.googleapis.com dataflow.googleapis.com logging.googleapis.com bigquery.googleapis.com pubsub.googleapis.com storage.googleapis.com cloudresourcemanager.googleapis.com cloudscheduler.googleapis.com

14. If you're using a local shell, then create local authentication credentials for your user account:

    gcloud auth application-default login

    You don't need to do this if you're using Cloud Shell.

15. Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/iam.serviceAccountUser

    gcloud projects add-iam-policy-binding PROJECT_ID --member="USER_IDENTIFIER" --role=ROLE

    • Replace PROJECT_ID with your project ID.

    • Replace USER_IDENTIFIER with the identifier for your user account. For example, user:[e-mail].

    • Replace ROLE with each individual role.

16. 向您的 Compute Engine 默认服务账号授予角色。对以下每个 IAM 角色运行以下命令一次：

    • roles/dataflow.admin
    • roles/dataflow.worker
    • roles/storage.admin
    • roles/pubsub.editor
    • roles/bigquery.dataEditor

    gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:[e-mail]" --role=SERVICE_ACCOUNT_ROLE

    替换以下内容：

    • PROJECT_ID：您的项目 ID。
    • PROJECT_NUMBER：您的项目编号。 如需查找您的项目编号，请使用 gcloud projects describe 命令。
    • SERVICE_ACCOUNT_ROLE：每个角色。

创建示例来源和接收器

本部分介绍如何创建以下内容：

• 使用 Pub/Sub 的流式处理数据源
• 将数据加载到 BigQuery 中的数据集

创建 Cloud Storage 存储桶

首先使用 Google Cloud 控制台或 Google Cloud CLI 创建 Cloud Storage 存储桶。Dataflow 流水线将此存储桶用作临时存储位置。

控制台

1. 在 Google Cloud 控制台中，进入 Cloud Storage 存储桶页面。

   进入“存储桶”

2. 点击创建存储桶。

3. 在创建存储桶页面上的指定存储桶的名称部分，输入符合存储桶命名要求的名称。Cloud Storage 存储桶名称必须是全局唯一的。 请勿选择其他选项。

4. 点击创建。

gcloud

使用 gcloud storage buckets create 命令：

gcloud storage buckets create gs://BUCKET_NAME

将 BUCKET_NAME 替换为符合存储桶命名要求的 Cloud Storage 存储桶的名称。Cloud Storage 存储桶名称必须是全局唯一的。

创建 Pub/Sub 主题和订阅

创建 Pub/Sub 主题，然后创建对该主题的订阅。

控制台

如需创建主题，请完成以下步骤。

1. 在 Google Cloud 控制台中，转到 Pub/Sub 主题页面。

   转到“主题”

2. 点击创建主题。

3. 在主题 ID 字段中，输入主题 ID。 如需了解如何命名主题，请参阅主题或订阅命名指导。

4. 保留添加默认订阅选项。请勿选择其他选项。

5. 点击创建主题。

gcloud

如需创建主题，请运行 gcloud pubsub topics create 命令。如需了解如何命名订阅，请参阅主题或订阅命名指南。

gcloud pubsub topics create TOPIC_ID

将 TOPIC_ID 替换为您的 Pub/Sub 主题的名称。

如需创建对主题的订阅，请运行 gcloud pubsub subscriptions create 命令：

gcloud pubsub subscriptions create --topic TOPIC_ID SUBSCRIPTION_ID

将 SUBSCRIPTION_ID 替换为您的 Pub/Sub 订阅的名称。

创建并运行 Cloud Scheduler 作业

创建并运行两个 Cloud Scheduler 作业，一个发布正面评分，另一个将负面评分发布到 Pub/Sub 主题。

控制台

创建 Cloud Scheduler 作业以获取正面评分。

1. 访问控制台中的 Cloud Scheduler 页面：

   转到 Cloud Scheduler

2. 点击创建作业按钮。

3. 输入名称 positive-ratings-publisher。

4. 选择靠近您在本教程中运行命令的位置的 Dataflow 区域。REGION 变量的值必须是有效的区域名称。如需详细了解区域和位置，请参阅 Dataflow 位置。

5. 使用 unix-cron 格式 * * * * * 指定作业的频率。

   如需了解详情，请参阅配置 Cron 作业时间表。

6. 选择您所在的时区。

7. 点击继续。

8. 在目标列表中，选择 Pub/Sub。

9. 从列表中选择您的主题名称。

10. 添加以下消息字符串以发送到您的目标： {"url": "https://beam.apache.org/", "review": "positive"}

11. 点击创建。

您现在将获得每分钟向您的 Pub/Sub 主题发送一条包含正面评分的消息的 cron 作业。该主题的 Cloud Functions 函数订阅。

创建 Cloud Scheduler 作业以发布负面评分。

1. 在控制台的 Cloud Scheduler 页面上，点击创建作业按钮。

2. 输入名称 negative-ratings-publisher。

3. 选择要在其中运行作业的区域。

4. 使用 unix-cron 格式 */2 * * * * 指定作业的频率。

   如需了解详情，请参阅配置 Cron 作业时间表。

5. 选择您所在的时区。

6. 点击继续。

7. 在目标列表中，选择 Pub/Sub。

8. 从列表中选择您的主题名称。

9. 添加以下消息字符串以发送到您的目标： {"url": "https://beam.apache.org/", "review": "negative"}

10. 点击创建。

您现在将获得每两分钟向 Pub/Sub 主题发送一条具有负面评分的消息的 Cron 作业。该主题的 Cloud Functions 函数订阅。

gcloud

1. 如需为本教程创建 Cloud Scheduler 作业，请使用 gcloud scheduler jobs create 命令。此步骤创建一个发布者以发布“正面评分”，该发布者每分钟发布一条消息。

   gcloud scheduler jobs create pubsub positive-ratings-publisher \
     --schedule="* * * * *" \
     --location=DATAFLOW_REGION \
     --topic="TOPIC_ID" \
     --message-body='{"url": "https://beam.apache.org/", "review": "positive"}'
   

   将 DATAFLOW_REGION 替换为要在其中部署 Dataflow 作业的区域。选择靠近您在本教程中运行命令的位置的 Dataflow 区域。REGION 变量的值必须是有效的区域名称。

2. 如需启动 Cloud Scheduler 作业，请使用 gcloud scheduler jobs run 命令。

   gcloud scheduler jobs run --location=DATAFLOW_REGION positive-ratings-publisher
   

3. 再创建并运行一个类似的“负面评分”发布者，该发布者每两分钟发布一条消息。此步骤会创建一个发布者以发布“负面评分”，该发布者每两分钟发布一条消息。

   gcloud scheduler jobs create pubsub negative-ratings-publisher \
     --schedule="*/2 * * * *" \
     --location=DATAFLOW_REGION  \
     --topic="TOPIC_ID" \
     --message-body='{"url": "https://beam.apache.org/", "review": "negative"}'
   

4. 启动第二个 Cloud Scheduler 作业。

   gcloud scheduler jobs run --location=DATAFLOW_REGION negative-ratings-publisher
   

创建 BigQuery 数据集

为您的 Pub/Sub 主题创建具有适当架构的 BigQuery 数据集和表。

控制台

创建 BigQuery 数据集。

1. 在 Google Cloud 控制台中打开 BigQuery 页面。

   转到 BigQuery 页面

2. 在探索器面板中，选择您要在其中创建数据集的项目。

3. 展开 操作选项，然后点击创建数据集。

4. 在创建数据集页面中执行以下操作：

   • 在数据集 ID 部分，输入 tutorial_dataset。

   • 对于数据位置，为数据集选择一个地理位置。创建数据集后，就无法再更改此位置。

     注意：如果您选择 EU 或欧盟的区域作为数据集位置，则核心 Dataflow 客户数据会驻留在欧盟。核心 Dataflow 客户数据在服务专用条款中定义。

   • 请勿选择其他选项。

   • 点击创建数据集。

创建具有架构的 BigQuery 表。

1. 在浏览器面板中，展开您的项目并选择您的 tutorial_dataset 数据集。

2. 展开 操作选项，然后点击打开。

3. 在详情面板中，点击创建表 。

4. 在创建表页面的来源部分，选择空白表。

5. 在创建表页面的目标部分，执行以下操作：

   • 验证数据集名称是否设置为 tutorial_dataset。
   • 在表名称字段中，输入 tutorial。
   • 确认表类型设置为原生表。

6. 在 Schema 部分中，输入架构定义。点击以文字形式修改，并以 JSON 数组格式输入以下表架构：

   [
     {
       "mode": "NULLABLE",
       "name": "url",
       "type": "STRING"
     },
     {
       "mode": "NULLABLE",
       "name": "review",
       "type": "STRING"
     }
   ]
   

7. 对于分区和聚簇设置，保留默认值 No partitioning。

8. 对于高级选项部分的加密，保留默认值 Google-managed key。默认情况下，Dataflow 会对以静态方式存储的客户内容进行加密。

9. 点击创建表。

gcloud

使用 bq mk 命令创建数据集。

bq --location=DATAFLOW_REGION mk \
PROJECT_ID:tutorial_dataset

将 PROJECT_ID 替换为项目的项目 ID。

搭配使用 bq mk 命令和 --table 或 -t 标志以在数据集中创建表。

bq mk \
    --table \
    PROJECT_ID:tutorial_dataset.tutorial \
    url:STRING,review:STRING

创建用户定义的函数 (UDF)

您可以视需要创建 JavaScript UDF 来扩展 Google 提供的 Pub/Sub Subscription to BigQuery 模板。借助 UDF，您可以定义模板中不存在的数据转换，并将其注入到模板中。

注意：Dataflow UDF 采用 JavaScript 编写，且必须可以在 Nashorn JavaScript 引擎中执行。

以下 UDF 会验证传入评分的网址。没有网址或网址错误的评分将转发到同一项目和数据集中带有 _error_records 后缀的另一个输出表（也称为死信表）。

JavaScript

/**
 * User-defined function (UDF) to transform events
 * as part of a Dataflow template job.
 *
 * @param {string} inJson input Pub/Sub JSON message (stringified)
 */
 function process(inJson) {
    const obj = JSON.parse(inJson);
    const includePubsubMessage = obj.data && obj.attributes;
    const data = includePubsubMessage ? obj.data : obj;

    if (!data.hasOwnProperty('url')) {
      throw new Error("No url found");
    } else if (data.url !== "https://beam.apache.org/") {
      throw new Error("Unrecognized url");
    }

    return JSON.stringify(obj);
  }

将此 JavaScript 代码段保存到之前创建的 Cloud Storage 存储桶中。

运行流水线

使用 Google 提供的 Pub/Sub Subscription to BigQuery 模板运行流处理流水线。该流水线从 Pub/Sub 主题获取传入数据，并将该数据输出到 BigQuery 数据集。

控制台

1. 在 Google Cloud 控制台中，转到 Dataflow 作业页面。

   转到作业

2. 点击基于模板创建作业。

3. 为 Dataflow 作业输入作业名称。

4. 在区域端点部分中，为 Dataflow 作业选择一个区域。

5. 在 Dataflow 模板部分中，选择 Pub/Sub Subscription to BigQuery 模板。

6. 对于 BigQuery 输出表，输入以下内容：

   PROJECT_ID:tutorial_dataset.tutorial
   

7. 在 Pub/Sub input subscription 部分中，输入以下内容：

   projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID
   

   将 PROJECT_ID 替换为在其中创建了 BigQuery 数据集的项目的 ID，将 SUBSCRIPTION_ID 替换为 Pub/Sub 订阅的名称。

8. 对于临时位置，输入以下内容：

   gs://BUCKET_NAME/temp/
   

   将 BUCKET_NAME 替换为您的 Cloud Storage 存储桶的名称。temp 文件夹存储临时文件，如暂存的流水线作业。

9. 可选：如需为作业添加 UDF，请展开可选参数。

   1. 在 Cloud Storage 中的 JavaScript UDF 路径部分，输入以下内容：

      gs://BUCKET_NAME/dataflow_udf_transform.js
      

   2. 在 JavaScript UDF 名称部分，输入以下内容：

      process
      

10. 点击运行作业。

如需检查模板是否能够将消息转发到死信表，请发布一些没有网址或网址错误的评分。

1. 转到 Pub/Sub 主题页面。

2. 点击您的 TOPIC_ID。

3. 转到消息部分。

4. 点击发布消息。

5. 在消息正文部分输入一些没有网址或网址错误的评分。例如：

   {"url": "https://beam.apache.org/documentation/sdks/java/", "review": "positive"}
   

6. 点击发布。

gcloud

注意：如需使用 Google Cloud CLI 运行经典模板，您必须拥有 gcloud CLI 138.0.0 版或更高版本。

如需在 shell 或终端中运行模板，请使用 gcloud dataflow jobs run 命令。

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-DATAFLOW_REGION/latest/PubSub_Subscription_to_BigQuery \
    --region DATAFLOW_REGION \
    --staging-location gs://BUCKET_NAME/temp \
    --parameters \
inputSubscription=projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID,\
outputTableSpec=PROJECT_ID:tutorial_dataset.tutorial

将 JOB_NAME 替换为您选择的唯一名称。

（可选）如需运行包含 UDF 的模板，请使用以下命令：

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-DATAFLOW_REGION/latest/PubSub_Subscription_to_BigQuery \
    --region DATAFLOW_REGION \
    --staging-location gs://BUCKET_NAME/temp \
    --parameters \
inputSubscription=projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID,\
outputTableSpec=PROJECT_ID:tutorial_dataset.tutorial,\
javascriptTextTransformGcsPath=gs://BUCKET_NAME/dataflow_udf_transform.js,\
javascriptTextTransformFunctionName=process

如需检查模板是否能够将消息转发到死信表，请发布一些没有网址或网址错误的评分。例如：

gcloud pubsub topics publish TOPIC_ID \
  --message='{"url": "https://beam.apache.org/documentation/sdks/java/", "review": "positive"}'

查看结果

查看写入 BigQuery 表的数据。

控制台

1. 在 Google Cloud 控制台中，转到 BigQuery 页面。
   转到 BigQuery 页面

2. 在查询编辑器中，运行以下查询：

   SELECT * FROM `PROJECT_ID.tutorial_dataset.tutorial`
   LIMIT 1000
   

   数据最多可能需要一分钟才会开始显示在表中。

   查询将返回在过去 24 小时内添加到表中的行。您还可以使用标准 SQL 运行查询。

   如果您希望将某些错误记录写入死信表，请在查询中使用表名称 tutorial_error_records。例如：

   SELECT * FROM `PROJECT_ID.tutorial_dataset.tutorial_error_records`
   LIMIT 1000
   

gcloud

通过运行以下查询来查看 BigQuery 中的结果：

bq query --use_legacy_sql=false 'SELECT * FROM `'"PROJECT_ID.tutorial_dataset.tutorial"'`'

在此流水线的运行期间，您可以看到 BigQuery 表中每分钟都在附加新行。

如果您希望将某些错误记录写入死信表，请在查询中使用表名称 tutorial_error_records。例如：

SELECT * FROM `PROJECT_ID.tutorial_dataset.tutorial_error_records`
LIMIT 1000

清理

为避免因本教程中使用的资源导致您的 Google Cloud 账号产生费用，请删除包含这些资源的项目，或者保留项目但删除各个资源。

删除项目

为了避免产生费用，最简单的方法是删除您为本教程创建的 Google Cloud 项目。

控制台

注意：删除项目会产生以下影响：

• 项目中的所有内容都会被删除。如果您将现有项目用于本文档中的任务，则删除该项目后，还将删除您已在该项目中完成的任何其他工作。
• 自定义项目 ID 丢失。创建此项目时，您可能创建了要在将来使用的自定义项目 ID。要保留使用该项目 ID 的网址（例如 appspot.com 网址），请删除项目内的所选资源，而不是删除整个项目。

如果您打算探索多个架构、教程或快速入门，则重复使用项目可以帮助您避免超出项目配额限制。

1. In the Google Cloud console, go to the Manage resources page.

   Go to Manage resources

2. In the project list, select the project that you want to delete, and then click Delete.
3. In the dialog, type the project ID, and then click Shut down to delete the project.

gcloud

注意：删除项目会产生以下影响：

• 项目中的所有内容都会被删除。如果您将现有项目用于本文档中的任务，则删除该项目后，还将删除您已在该项目中完成的任何其他工作。
• 自定义项目 ID 丢失。创建此项目时，您可能创建了要在将来使用的自定义项目 ID。要保留使用该项目 ID 的网址（例如 appspot.com 网址），请删除项目内的所选资源，而不是删除整个项目。

如果您打算探索多个架构、教程或快速入门，则重复使用项目可以帮助您避免超出项目配额限制。

Delete a Google Cloud project:

gcloud projects delete PROJECT_ID

逐个删除资源

如果您希望以后重复使用该项目，可以保留该项目，但删除在本教程中创建的资源。

停止 Dataflow 流水线

控制台

1. 在 Google Cloud 控制台中，转到 Dataflow 作业页面。

   转到作业

2. 点击要停止的作业。

   要停止作业，作业状态必须为正在运行。

3. 在作业详情页面上，点击停止。

4. 点击取消。

5. 要确认您的选择，请点击停止作业。

gcloud

如需取消 Dataflow 作业，请使用 gcloud dataflow jobs 命令。

gcloud dataflow jobs list \
  --filter 'NAME=JOB_NAME AND STATE=Running' \
  --format 'value(JOB_ID)' \
  --region "DATAFLOW_REGION" \
  | xargs gcloud dataflow jobs cancel --region "DATAFLOW_REGION"

清理 Google Cloud 项目资源

控制台

1. 删除 Cloud Scheduler 作业。

   1. 在 Google Cloud 控制台中，转到 Cloud Scheduler 页面。

      转到 Cloud Scheduler

   2. 选择您的作业。

   3. 点击页面顶部的删除按钮并确认删除操作。

2. 删除 Pub/Sub 主题和订阅。

   1. 进入 Google Cloud 控制台中的 Pub/Sub 主题页面。

      打开“主题”。

   2. 选择您创建的主题。

   3. 点击删除以永久删除该主题。

   4. 进入 Google Cloud 控制台中的 Pub/Sub 订阅页面。

      前往订阅页面

   5. 选择使用您的主题创建的订阅。

   6. 点击删除以永久删除该订阅。

3. 删除 BigQuery 表和数据集。

   1. 在 Google Cloud 控制台中，转到 BigQuery 页面。

      前往 BigQuery

   2. 在探索器面板中，展开您的项目。

   3. 在要删除的数据集旁边，点击 查看操作，然后点击删除。

4. 删除 Cloud Storage 存储桶。

   1. 在 Google Cloud 控制台中，进入 Cloud Storage 存储桶页面。

      进入“存储桶”

   2. 选择要删除的存储t桶，点击 删除，然后按照说明操作。

gcloud

1. 如需删除 Cloud Scheduler 作业，请使用 gcloud scheduler jobs delete 命令。

   gcloud scheduler jobs delete negative-ratings-publisher --location=DATAFLOW_REGION
   

   gcloud scheduler jobs delete positive-ratings-publisher --location=DATAFLOW_REGION
   

2. 如需删除 Pub/Sub 订阅和主题，请使用 gcloud pubsub subscriptions delete 和 gcloud pubsub topics delete 命令。

   gcloud pubsub subscriptions delete SUBSCRIPTION_ID
   gcloud pubsub topics delete TOPIC_ID
   

3. 如需删除 BigQuery 表，请使用 bq rm 命令。

   bq rm -f -t PROJECT_ID:tutorial_dataset.tutorial
   

4. 删除 BigQuery 数据集。单独的数据集不会产生任何费用。

   注意：以下命令还会删除该数据集中的所有表。表和数据无法恢复。

   bq rm -r -f -d PROJECT_ID:tutorial_dataset
   

5. 如需删除 Cloud Storage 存储桶，请使用 gcloud storage rm 命令。单独的存储桶不会产生任何费用。

   注意：以下命令还会删除该存储桶中的所有对象。无法恢复这些对象。

   gcloud storage rm gs://BUCKET_NAME --recursive
   

撤销凭据

控制台

如果您保留项目，请撤消授予 Compute Engine 默认服务账号的角色。

1. 在 Google Cloud 控制台中，转到 IAM 页面。

转到 IAM

1. 选择一个项目、文件夹或组织。

2. 找到包含要撤消其访问权限的主账号的行。 在该行中，点击 修改主账号。

3. 点击要撤消的每个角色对应的删除 按钮，然后点击保存。

gcloud

• 如果您保留项目，请撤消授予 Compute Engine 默认服务账号的角色。对以下每个 IAM 角色运行以下命令一次：
  • roles/dataflow.admin
  • roles/dataflow.worker
  • roles/storage.admin
  • roles/pubsub.editor
  • roles/bigquery.dataEditor

    gcloud projects remove-iam-policy-binding <var>PROJECT_ID</var> \
    --member=serviceAccount:<var>PROJECT_NUMBER</var>[e-mail] \
    --role=<var>ROLE</var>
  

• Optional: Revoke the authentication credentials that you created, and delete the local credential file.

  gcloud auth application-default revoke

• Optional: Revoke credentials from the gcloud CLI.

  gcloud auth revoke

后续步骤

• 使用 UDF 扩展 Dataflow 模板。
• 详细了解如何使用 Dataflow 模板。
• 查看 Google 提供的所有模板。
• 了解如何使用 Pub/Sub 创建和使用主题以及如何创建拉取订阅。
• 详细了解如何使用 Cloud Scheduler 安排和运行 Cron 作业。
• 了解如何使用 BigQuery 创建数据集。
• 了解 Pub/Sub 订阅。
• 探索有关 Google Cloud 的参考架构、图表和最佳实践。查看我们的 Cloud Architecture Center。