实现 Datastream 和 Dataflow 以进行分析

---

Datastream 支持将数据从 Oracle、MySQL 和 PostgreSQL 数据库直接流式传输到 BigQuery 数据集。但是，如果您需要更好地控制流处理逻辑（例如数据转换或手动设置逻辑主键），可以将 Datastream 与 Dataflow 作业模板集成。

本教程介绍了如何使用 Dataflow 作业模板在 BigQuery 中流式传输最新具体化视图以进行分析，从而与 Dataflow 集成。

对于具有许多独立数据源的组织而言，对整个组织中企业数据的访问（尤其是实时访问）可能会受到限制，速度也很慢。这会限制组织的内省能力。

Datastream 可让您近乎实时地访问来自各种本地和云端数据源的数据更改。Datastream 提供一种设置体验，让您无需对流式数据进行太多配置；Datastream 会为您完成这项工作。Datastream 还具有统一的使用 API，可让您的组织访问最新的可用企业数据以构建集成方案。

其中一种情况是将数据从源数据库传输到云端存储服务或消息传递队列。在 Datastream 流式传输数据后，数据将转换为其他应用和服务可以读取的形式。在本教程中，Dataflow 是一项网络服务，用于与存储服务或消息传递队列通信，以在 Google Cloud 上捕获和处理数据。

了解如何使用 Datastream 将源 MySQL 数据库中的更改（插入、更新或删除的数据）流式传输到 Cloud Storage 存储桶中的文件夹。然后，配置 Cloud Storage 存储桶以发送通知，以供 Dataflow 了解包含 Datastream 从源数据库流式传输的数据更改的任何新文件。然后，Dataflow 作业会处理文件并将更改传输到 BigQuery。

目标

在本教程中，您将执行以下操作：

• 在 Cloud Storage 中创建存储桶。这是目标存储桶，Datastream 会将架构、表和数据从源 MySQL 数据库流式传输到其中。
• 为 Cloud Storage 存储桶启用 Pub/Sub 通知。这样即可配置存储桶以发送通知，以供 Dataflow 了解任何已可供处理的新文件。这些文件包含 Dataflow 从源数据库流式传输到存储桶的数据更改。
• 在 BigQuery 中创建数据集。BigQuery 使用数据集来包含它从 Dataflow 接收的数据。此数据表示 Datastream 流式传输到 Cloud Storage 存储桶中的源数据库更改。
• 创建和管理用于 Cloud Storage 中源数据库和目标存储桶的连接配置文件。Datastream 中的数据流使用连接配置文件中的信息将源数据库中的数据流式传输到存储桶。
• 创建并启动数据流。此数据流会将源数据库中的数据、架构和表转移到存储桶中。
• 验证 Datastream 是否将与源数据库的架构关联的数据和表转移到存储桶中。
• 在 Dataflow 中创建作业。在 Datastream 将数据更改从源数据库流式传输到 Cloud Storage 存储桶后，系统会向 Dataflow 发送关于包含这些更改的新文件的通知。Dataflow 作业会处理这些文件并将更改转移到 BigQuery。
• 验证 Dataflow 是否处理包含与此数据关联的更改的文件并将更改转移到 BigQuery。因此，Datastream 和 BigQuery 之间具有端到端集成。
• 清理您在 Datastream、Cloud Storage、Pub/Sub、Dataflow 和 BigQuery 上创建的资源，以免这些资源占用配额，日后产生费用。

费用

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

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

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

准备工作

1. 登录您的 Google Cloud 账号。如果您是 Google Cloud 新手，请创建一个账号来评估我们的产品在实际场景中的表现。新客户还可获享 $300 赠金，用于运行、测试和部署工作负载。

2. 在 Google Cloud Console 中的项目选择器页面上，选择或创建一个 Google Cloud 项目。

   注意：如果您不打算保留在此过程中创建的资源，请创建新的项目，而不要选择现有的项目。完成本教程介绍的步骤后，您可以删除所创建的项目，并移除与该项目关联的所有资源。

   转到“项目选择器”

3. 确保您的 Google Cloud 项目已启用结算功能。

4. 在 Google Cloud Console 中的项目选择器页面上，选择或创建一个 Google Cloud 项目。

   注意：如果您不打算保留在此过程中创建的资源，请创建新的项目，而不要选择现有的项目。完成本教程介绍的步骤后，您可以删除所创建的项目，并移除与该项目关联的所有资源。

   转到“项目选择器”

5. 确保您的 Google Cloud 项目已启用结算功能。

6. 启用 Datastream API。

   启用 API

7. 请确保您已向用户账号授予 Datastream Admin 角色。

   转到 IAM 页面

8. 确保您拥有 Datastream 可以访问的源 MySQL 数据库。此外，验证数据库中是否有数据、表和架构。
9. 配置 MySQL 数据库以允许来自 Datastream 公共 IP 地址的传入连接。如需查看所有 Datastream 区域及其关联的公共 IP 地址的列表，请参阅 IP 许可名单和区域。
10. 为源数据库设置变更数据捕获 (CDC)。如需了解详情，请参阅配置源 MySQL 数据库。

11. 确保您满足所有前提条件，以便为 Cloud Storage 启用 Pub/Sub 通知。

    在本教程中，您将在 Cloud Storage 中创建目标存储桶并为其启用 Pub/Sub 通知。这样设置后，Dataflow 就可以接收通知来了解 Datastream 写入存储桶的新文件。这些文件包含 Dataflow 从源数据库流式传输到存储桶的数据更改。

使用要求

Datastream 提供各种来源选项、目标选项和网络连接方法。

在本教程中，假设您使用独立的 MySQL 数据库和目标 Cloud Storage 服务。对于源数据库，您应该能够将网络配置为添加入站防火墙规则。源数据库可以位于本地，也可以位于云提供商。对于 Cloud Storage 目标位置，无需配置连接。

由于我们无法获知您环境的具体细节，因此我们无法提供网络配置的详细步骤。

在本教程中，请选择 IP 许可名单作为网络连接方法。IP 许可名单是一项安全功能，通常用于仅限受信任的用户访问您的源数据库中的数据并对这些访问进行控制。您可以使用 IP 许可名单创建受信任的 IP 地址或 IP 地址范围列表，您的用户和其他 Google Cloud 服务（如 Datastream）可通过这些地址访问此数据。要使用 IP 许可名单，您必须向来自 Datastream 的传入连接开放源数据库或防火墙。

在 Cloud Storage 中创建存储桶

在 Cloud Storage 中创建目标存储桶，Datastream 会将源 MySQL 数据库中的架构、表和数据流式传输到其中。

1. 在 Google Cloud 控制台中，前往 Cloud Storage 的浏览器页面。

   转到“浏览器”页面

2. 点击创建存储桶。此时将显示创建存储桶页面。

3. 在指定存储桶的名称区域的文本字段中，输入 my-integration-bucket，然后点击继续。

4. 对于页面每个剩余区域，请接受默认设置。在每个区域结束时，点击继续。

5. 点击创建。

为 Cloud Storage 存储桶启用 Pub/Sub 通知

在本部分中，您将为创建的 Cloud Storage 存储桶启用 Pub/Sub 通知。这样，您就可以将存储桶配置为在 Datastream 向存储桶写入任何新文件时通知 Dataflow。这些文件包含 Datastream 从源 MySQL 数据库流式传输到存储桶的数据的更改。

在为存储桶启用通知之前，请确保满足所有前提条件。

1. 访问您创建的 Cloud Storage 存储桶。此时将显示存储桶详情页面。

2. 点击 激活 Cloud Shell。

3. 在提示符处，输入以下命令：

   gsutil notification create -t my_integration_notifs -f json -p integration/tutorial/ gs://my-integration-bucket

   通过输入此命令，您将在 Pub/Sub 中创建 my_integration_notifs 主题。您还将 Pub/Sub 通知配置为仅监听您在为 Cloud Storage 创建目标连接配置文件步骤中定义的 Datastream 输出路径。 稍后，您将在 Pub/Sub 中配置此主题，让系统向 Dataflow 发送关于 Datastream 流式传输到 Cloud Storage 存储桶中的数据更改的通知。

   Pub/Sub 主题会捕获对此数据所做的任何更改。此主题的所有订阅者（例如 Dataflow）都会收到此信息。

4. 可选：如果出现为 Cloud Shell 授权窗口，请点击授权。

5. 验证您是否看到以下代码行：

   Created Cloud Pub/Sub topic projects/project-name/topics/my_integration_notifs
   Created notification config projects/_/buckets/my-integration-bucket/notificationConfigs/1
   

   project-name 是您的 Google Cloud 项目名称的占位符。

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

   转到“主题”页面

7. 点击您创建的 my_integration_notifs 主题。

8. 在 my_integration_notifs 页面，滚动到页面底部。验证订阅标签页是否处于活动状态，以及是否显示没有可显示的订阅消息。

   您将为 my_integration_notifs 主题创建订阅。订阅此订阅的应用（例如 Dataflow）可以获取主题中的信息。此信息与 Datastream 流式传输到 Cloud Storage 存储桶的源数据库数据更改相关联。

9. 点击创建订阅。

10. 在显示的菜单中，选择创建订阅。

11. 在为主题添加订阅页面中执行以下操作：

    1. 在订阅 ID 字段中，输入 my_integration_notifs_sub。
    2. 将确认时限值设置为 120 秒。这样可让 Dataflow 有足够的时间来确认已处理的文件，并有助于提高 Dataflow 作业的整体性能。如需详细了解 Pub/Sub 订阅属性，请参阅订阅属性。
    3. 保留页面上的所有其他默认值。
    4. 点击创建。

在本教程后面部分，您将创建 Dataflow 作业。创建此作业的过程中，您要将 Dataflow 指定为 my_integration_notifs_sub 订阅的订阅者。指定后，Dataflow 可以接收通知来了解 Datastream 写入 Cloud Storage 的新文件、处理这些文件并将数据更改转移到 BigQuery。

在 BigQuery 中创建数据集

在本部分中，您将在 BigQuery 中创建数据集。BigQuery 使用数据集来包含它从 Dataflow 接收的数据。此数据表示 Datastream 将流式传输到 Cloud Storage 存储桶的源 MySQL 数据库中的更改。

1. 在 Google Cloud 控制台中，转到 BigQuery 的 SQL 工作区页面。

   转到“SQL 工作区”页面

2. 在探索器窗格中，点击 Google Cloud 项目名称旁边的 查看操作。

3. 在显示的菜单中，选择创建数据集。

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

   1. 在数据集 ID 字段中，输入数据集的 ID。在本教程中，在字段中输入 My_integration_dataset_log。
   2. 保留窗口中的所有其他默认值。
   3. 点击创建数据集。
   Dataflow 使用 My_integration_dataset_log 数据集来暂存从 Datastream 收到的数据更改。

5. 在探索器窗格中，点击 Google Cloud 项目名称旁边的 展开节点，然后验证您能否看到您创建的数据集。

6. 按照此过程中的步骤创建第二个数据集：My_integration_dataset_final。

   在 My_integration_dataset_log 数据集中暂存的更改将合并到 My_integration_dataset_Final 数据集中，以创建源数据库中表的一对一副本。

7. 在每个数据集旁边，展开 展开节点。

8. 验证每个数据集是否为空。

在 Datastream 将数据更改从源数据库流式传输到您的 Cloud Storage 存储桶后，Dataflow 作业会处理包含这些更改的文件，并将更改传输到 BigQuery 数据集内。

在 Datastream 中创建连接配置文件

在本部分中，您将在 Datastream 中为源数据库和目标位置创建连接配置文件。在创建连接配置文件时，您选择 MySQL 作为来源连接配置文件的配置文件类型，选择 Cloud Storage 作为目标连接配置文件的配置文件类型。

Datastream 使用连接配置文件中定义的信息连接到源数据库和目标数据库，以便将数据从源数据库流式传输到 Cloud Storage 中的目标存储桶。

为 MySQL 数据库创建源连接配置文件

1. 在 Google Cloud 控制台中，前往 Datastream 的连接配置文件页面。

   转到“连接配置文件”页面

2. 点击 Create profile（创建配置文件）。

3. 如需为 MySQL 数据库创建源连接配置文件，请在创建连接配置文件页面中，点击 MySQL 配置文件类型。

4. 在创建 MySQL 配置文件页面的定义连接设置部分中，提供以下信息：

   • 在连接配置文件名称字段中，输入 My Source Connection Profile。
   • 保留自动生成的连接配置文件 ID。

   • 选择您要存储连接配置文件的区域。

     与所有资源一样，连接配置文件也保存在地区中，并且数据流只能使用与该数据流存储在同一地区的连接配置文件。地区选择不会影响 Datastream 是否可连接到来源或目标位置，但如果该地区发生停机，则可能会影响可用性。

   • 输入连接详情：

     • 在主机名或 IP 字段中，输入 Datastream 可用于连接到源数据库的主机名或公共 IP 地址。您提供公共 IP 地址是因为您将 IP 许可名单用作本教程的网络连接方法。
     • 在端口字段中，输入为源数据库预留的端口号。对于 MySQL 数据库，默认端口通常为 3306。
     • 输入用户名和密码，对源数据库进行身份验证。

5. 在定义连接设置部分中，点击继续。创建 MySQL 配置文件页面的确保与源的连接安全部分处于活跃状态。

6. 从加密类型菜单中，选择无。如需详细了解此菜单，请参阅为 MySQL 数据库创建连接配置文件。

7. 在保护与来源的连接部分，点击继续。创建 MySQL 配置文件页面的定义连接方法部分处于活跃状态。

8. 在连接方法下拉列表中，选择要在 Datastream 和源数据库之间建立连接的网络方法。在本教程中，请选择 IP 许可名单作为连接方法。

9. 配置源数据库以允许来自显示的 Datastream 公共 IP 地址的传入连接。

10. 在定义连接方法部分中，点击继续。创建 MySQL 配置文件页面的测试连接配置文件部分处于活跃状态。

11. 点击运行测试以验证源数据库和 Datastream 是否可以相互通信。

12. 验证您是否看到测试已通过状态。

    如果测试失败，您可以在流程的相应部分解决问题，然后返回重新测试。请参阅诊断问题页面，了解问题排查步骤。

13. 点击创建。

为 Cloud Storage 创建目标连接配置文件

1. 在 Google Cloud 控制台中，前往 Datastream 的连接配置文件页面。

   转到“连接配置文件”页面

2. 点击 Create profile（创建配置文件）。

3. 如需为 Cloud Storage 创建目标连接配置文件，请在创建连接配置文件页面中，点击 Cloud Storage 配置文件类型。

4. 在创建 Cloud Storage 配置文件页面中，提供以下信息：

   • 在连接配置文件名称字段中，输入 My Destination Connection Profile。
   • 保留自动生成的连接配置文件 ID。
   • 选择您要存储连接配置文件的区域。

   • 在连接详情窗格中，点击浏览，选择您在本教程前面创建的 my-integration-bucket。这是 Datastream 将数据从源数据库转移到其中的存储桶。选择后，点击选择。

     您的存储桶会显示在连接详情窗格的存储桶名称字段中。

   • 在连接配置文件路径前缀字段中，提供在 Datastream 将数据流式传输至目的地时要附加到存储桶名称的路径的前缀。确保 Datastream 将数据写入存储分区内的路径，而不是存储桶根文件夹。在本教程中，请使用您在配置 Pub/Sub 通知时定义的路径。在字段中输入 /integration/tutorial。

5. 点击创建。

为 MySQL 数据库创建源连接配置文件和 Cloud Storage 目标连接配置文件后，您可以使用它们创建数据流。

在 Datastream 中创建数据流

在本部分中，您将创建一个数据流。此数据流使用连接配置文件中的信息将数据从源 MySQL 数据库转移到 Cloud Storage 中的目标存储桶。

Datastream 附带设置助理，以帮助您创建数据流。此助理由六个面板组成：“使用入门”“定义和测试来源”“配置来源”“定义目标”“配置目标”以及“查看并创建”。如需填充各个面板，请按照以下部分中的说明操作。

定义数据流的设置

1. 在 Google Cloud 控制台中，前往 Datastream 的数据流页面。

   转到“数据流”页面

2. 点击创建信息流。

3. 在创建数据流页面的定义数据流详情面板中提供以下信息：

   • 在数据流名称字段中，输入 My Stream。
   • 保留自动生成的数据流 ID。
   • 从地区菜单中，选择您在其中创建了来源和目标连接配置文件的地区。
   • 从来源类型菜单中，选择 MySQL 配置文件类型。
   • 从目标类型菜单中，选择 Cloud Storage 配置文件类型。

4. 查看自动生成的必要前提条件，以反映如何为数据流准备您的环境。这些前提条件可能包括如何配置源数据库以及如何将 Datastream 连接到 Cloud Storage 中的目标存储桶。

   如需详细了解这些前提条件，请参阅配置源 MySQL 数据库。

5. 点击继续。系统会显示创建数据流页面的定义 MySQL 连接配置文件面板。

指定有关来源连接配置文件的信息

在本部分中，您将选择为源数据库创建的连接配置文件（源连接配置文件）。在本教程中，名称为 My Source Connection Profile。

1. 从来源连接配置文件菜单中，选择 MySQL 数据库的来源连接配置文件。

2. 点击运行测试以验证源数据库和 Datastream 是否可以相互通信。

   如果测试失败，会显示与连接配置文件关联的问题。请参阅诊断问题页面，了解问题排查步骤。进行必要的更改以纠正问题，然后重新测试。

3. 点击继续。系统会显示创建数据流页面的配置数据流来源面板。

配置有关数据流的源数据库的信息

在本部分中，您将通过指定源数据库中 Datastream 使用的表和架构，为数据流配置有关源数据库的信息：

• 可以转移到目标位置。
• 无法转移到目标位置。

您还可以确定 Datastream 是回填历史数据、将正在进行的更改流式传输到目标位置，还是仅流式传输数据更改。

1. 使用要包含的对象菜单指定源数据库中 Datastream 可以转移到 Cloud Storage 目标存储桶中的文件夹的表和架构。只有当您的数据库中最多包含 5,000 个对象时，该菜单才会加载。

   在本教程中，您希望 Datastream 转移所有表和架构。因此，请从菜单中选择所有架构中的所有表。

2. 验证选择要排除的对象面板是否已设置为无。您不希望限制 Dataflow 将源数据库中的任何表和架构转移到 Cloud Storage 的操作。

3. 验证为历史数据选择回填模式 面板是否已设置为自动。除数据更改外，Datastream 会将所有现有数据从来源流式传输到目标位置。

4. 点击继续。系统显示创建数据流页面的定义 Cloud Storage 连接配置文件面板。

选择目标连接配置文件

在本部分中，您将选择为 Cloud Storage 创建的连接配置文件（目标连接配置文件）。在本教程中，该配置文件为 My Destination Connection Profile。

1. 从目标连接配置文件菜单中，选择 Cloud Storage 的目标连接配置文件。

2. 点击继续。系统会显示创建数据流页面的配置数据流目标位置面板。

配置有关数据流的目标位置的信息

在本部分中，您将配置流的目标存储桶的相关信息。此类信息包括：

• 写入 Cloud Storage 的文件的输出格式。
• 目标存储桶的文件夹，Datastream 会将架构、表和数据从源数据库转移到其中。

1. 在输出格式字段中，选择写入 Cloud Storage 的文件的格式。Datastream 支持两种输出格式：Avro 和 JSON。在本教程中，使用 Avro 文件格式。

   您无需填充数据流路径前缀字段。在此字段中，您可以为 Datastream 将数据传输到目的地时附加到存储桶名称的路径提供前缀。这是 Cloud Storage 存储桶的路径，Datastream 会将架构、表和数据从源 MySQL 数据库转移到该存储分区中。

   在为 Cloud Storage 创建目标连接配置文件时，您已经提供了 /integration/tutorial 的路径。

2. 点击继续。系统会显示创建数据流页面的审核数据流详情并创建面板。

创建数据流

1. 验证数据流的详细信息，以及数据流用于将数据从源 MySQL 数据库转移到 Cloud Storage 目标存储桶的源连接配置文件和目标连接配置文件。

2. 如需验证数据流，请点击运行验证。通过验证数据流，Datastream 会检查来源是否配置正确，验证数据流是否可以连接到来源和目标位置，并验证数据流的端到端配置。

   如果验证检查通过，那么系统会显示对勾标记图标。

   如果验证检查未通过，系统会显示一个感叹号图标和查看错误详情按钮。点击该按钮后，系统会显示一个弹出式窗口，其中说明了检查未通过的原因以及如何纠正问题。进行适当的更正后，点击重新验证。

   如需详细了解如何排查未通过的验证检查，请参阅诊断问题。

3. 所有验证检查均通过后，点击创建。

4. 在创建数据流？对话框中，点击创建。

启动数据流

在本教程中，您将分别创建和启动数据流，以防数据流创建过程造成源数据库负载增加。如需推迟该负载，您可以在不启动数据流的情况下创建数据流，然后在数据库可以处理负载时启动数据流。

通过启动数据流，Datastream 可以将数据、架构和表从源数据库转移到目标位置。

1. 在 Google Cloud 控制台中，前往 Datastream 的数据流页面。

   转到“数据流”页面

2. 选中要开始的直播旁边的复选框。在本教程中，这是我的数据流。

3. 点击启动。

4. 在对话框中，点击 Start。数据流的状态从 Not started 更改为 Starting，再更改为 Running。

   数据流大约需要 30 秒才能开始运行。需要启动后台资源，然后才能启动数据流。

启动数据流后，您可以验证 Dataflow 是否将数据从源数据库转移到目标位置。

验证数据流

在本部分中，您将确认 Dataflow 将数据从源 MySQL 数据库的所有表转移到 Cloud Storage 目标存储桶的 /integration/tutorial 文件夹中。在本教程中，您的存储桶的名称为 my-integration-bucket。

1. 在 Google Cloud 控制台中，前往 Datastream 的数据流页面。

   转到“数据流”页面

2. 点击您创建的数据流。在本教程中，这是我的数据流。

3. 在数据流详情页面中，点击 my-integration-bucket/integration/tutorial 链接。此链接显示在目标写入路径字段后面。Cloud Storage 的存储分区详情页面会在单独的标签页中打开。

   该链接包含存储桶的名称和存储桶所在的文件夹，Datastream 会将源数据库中的架构、表和数据转移到该存储桶中。在为 Cloud Storage 创建目标连接配置文件时，您已将此文件夹指定为 /integration/tutorial。

4. 验证您是否看到表示源数据库的表的文件夹。

5. 点击其中一个表文件夹，然后点击每个子文件夹，直到看到与表关联的数据。

   第一个文件夹是 [schema]_[table]，后跟文件夹表示 Datastream 将数据从源数据库转移到 Cloud Storage 中的目标存储桶的具体时间（年、月、日、小时和分钟）。

   每分钟创建一个文件夹（当要写入新数据时）。

   当文件大小达到 250 MB 或架构发生更改时，系统将创建一个新文件。如果表已分区，则系统会为每个分区创建文件。

创建 Dataflow 作业

在本部分中，您将在 Dataflow 中创建作业。当 Datastream 将数据更改从源 MySQL 数据库流式传输到您的 Cloud Storage 存储桶后，Pub/Sub 会向 Dataflow 发送关于包含这些更改的新文件的通知。Dataflow 作业会处理这些文件并将更改转移到 BigQuery。

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

   转到作业页面

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

3. 在从模板创建作业页面的作业名称字段中，输入您要创建的 Dataflow 作业的名称。在本教程中，在字段中输入 my-dataflow-integration-job。

4. 从区域端点菜单中，选择要存储作业的区域。此地区就是您为所创建的来源连接配置文件、目标连接配置文件和数据流选择的地区。

5. 从 Dataflow 模板菜单中，选择要用于创建作业的模板。在本教程中，请选择 Datastream to BigQuery。

   进行选择后，将显示与此模板相关的其他字段。

   Datastream to BigQuery 模板是一种流处理流水线，可读取 Datastream 数据并将其复制到 BigQuery 中。该模板使用 Pub/Sub 通知从 Cloud Storage 中读取数据，并将其复制到时间分区的 BigQuery 暂存表中。复制后，该模板会在 BigQuery 中执行 Merge，将所有变更数据捕获获 (CDC) 更改插入/更新到源表的副本中。

   为了最大限度降低频繁 Merge 操作产生的费用，我们建议首先采用初始频率，即每 6-12 小时操作一次。完成所有回填且顺利复制数据后，请将此值降低至所需的频率。

   如需详细了解 Datastream to BigQuery 模板，请参阅 Datastream to BigQuery (Stream)。

6. 在 Cloud Storage 中 Datastream 文件输出的文件位置字段中，输入包含 Cloud Storage 存储桶的名称的路径。在本教程中，请输入 gs://my-integration-bucket。

7. 在 Cloud Storage 通知政策中使用的 Pub/Sub 订阅字段中，输入包含 Pub/Sub 订阅名称的路径。在本教程中，请输入 projects/project-name/subscriptions/my_integration_notifs_sub。

   project-name 是您的 Google Cloud 项目名称的占位符。此外，您在本教程的为 Cloud Storage 存储桶启用 Pub/Sub 通知部分创建了 my_integration_notifs_sub 订阅。

8. 在 Datastream 输出文件格式 (avro/json) 字段中输入 avro，因为在本教程中，Avro 是 Datastream 写入 Cloud Storage 的文件的格式。

9. 在包含暂存表的数据集的名称或模板字段中，输入 My_integration_dataset_log，因为 Dataflow 使用此数据集来暂存其从 Datastream 收到的数据更改。

10. 在包含副本表的数据集模板字段中，输入 My_integration_dataset_final，因为这是数据集，其中暂存在 My_integration_dataset_log 数据集中的更改会在源数据库中创建表的一对一副本。

    在本教程的在 BigQuery 中创建数据集部分，您已创建了 My_integration_dataset_log 和 My_integration_dataset_ final 数据集。

11. 在死信队列目录字段中，输入包含 Cloud Storage 存储桶名称的路径和死信队列的文件夹。请勿使用根文件夹中的路径，并且该路径与 Datastream 写入数据的路径不同。Dataflow 无法传输到 BigQuery 的任何数据更改都会存储在队列中。您可以修复队列中的内容，以便 Dataflow 可以重新处理它。

    在本教程中，请在死信队列目录字段中输入 gs://my-integration-bucket/dlq（其中，dlq 是死信队列的文件夹）。

12. 点击运行作业。

Dataflow 作业可能需要几分钟才能运行完毕。需要运行后台资源。此外，作业必须先处理包含从 Cloud Storage 接收的数据更改的文件，然后才能将这些更改转移到 BigQuery 中。

验证集成

在本教程的验证数据流部分，您确认 Datastream 已将源 MySQL 数据库的所有表中的数据转移到 Cloud Storage 目标存储桶的 /integration/tutorial 文件夹中。

在本部分中，您将验证 Dataflow 是否处理包含与此数据关联的文件，以及是否将更改传输到 BigQuery。因此，Datastream 和 BigQuery 之间具有端到端集成。

1. 在 Google Cloud 控制台中，转到 BigQuery 的 SQL 工作区页面。

   转到“SQL 工作区”页面

2. 在探索器窗格中，展开 Google Cloud 项目名称旁边的节点。

3. 展开 My_integration_dataset_log 和 My_integration_dataset_Final 数据集旁边的节点。

4. 验证每个数据集现在是否包含数据。这会确认 Dataflow 已处理与 Datastream 流式传输到 Cloud Storage 中的数据相关联的文件，并且已将这些更改转移到 BigQuery 中。

清理

为避免因本教程中使用的资源导致您的 Google Cloud 帐号产生费用，请使用 Google Cloud 控制台执行以下操作：

• 删除您的项目、Datastream 数据流和 Datastream 连接配置文件。
• 停止 Dataflow 作业。
• 删除 BigQuery 数据集、Pub/Sub 主题和订阅以及 Cloud Storage 存储桶。

通过清理在 Datastream、Dataflow、BigQuery、Pub/Sub 和 Cloud Storage 上创建的资源，您可以防止这些资源占用配额，日后也无需为这些资源付费。

删除项目

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

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

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

如果您打算浏览多个教程和快速入门，重复使用项目可以帮助您避免超出项目配额上限。

1. 在 Google Cloud 控制台中，进入管理资源页面：

   转到“管理资源”页面

2. 在项目列表中，选择要删除的项目，然后点击删除。

3. 如需删除项目，请在对话框中输入项目 ID，然后点击关停。

删除数据流

1. 在 Google Cloud 控制台中，前往 Datastream 的数据流页面。

   转到“数据流”页面

2. 点击要删除的数据流。在本教程中，这是我的数据流。

3. 点击暂停。

4. 在对话框中，点击暂停。

5. 在数据流详情页面的数据流状态窗格中，验证数据流的状态是否为 Paused。

   请在删除数据流之前暂停数据流，以确保所有运行中数据都会从源数据库转移到目标数据库。

6. 点击删除。

7. 在对话框的文本字段中输入 Delete，然后点击 Delete。

删除连接配置文件

1. 在 Google Cloud 控制台中，前往 Datastream 的连接配置文件页面。

   转到“连接配置文件”页面

2. 选中要删除的每个连接配置文件对应的复选框：我的来源连接配置文件和我的目标连接配置文件。

3. 点击删除。

4. 在该对话框中，点击删除。

停止 Dataflow 作业

您不能在 Dataflow 中删除作业。但是，您可以停止作业，以便该作业不再处理包含流式传输到 Cloud Storage 存储桶的数据更改或将数据更改转移到 BigQuery 的文件。

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

   转到作业页面

2. 点击要停止的作业。在本教程中，这是 my-dataflow-integration-job。

3. 点击停止。

4. 在停止作业对话框中，选择排空选项，然后点击停止作业。

   Dataflow 会完成对所有剩余的缓冲数据的处理，以便所有运行中”数据在作业停止前转移到 BigQuery。

删除 BigQuery 数据集

1. 在 Google Cloud 控制台中，转到 BigQuery 的 SQL 工作区页面。

   转到“SQL 工作区”页面

2. 在探索器窗格中，展开 Google Cloud 项目名称旁边的节点。

3. 点击您在在 BigQuery 中创建数据集中创建的某个数据集右侧的查看操作按钮。该按钮形似垂直省略号。

   在本教程中，请点击 My_integration_dataset_log 右侧的查看操作按钮。

4. 从显示的下拉菜单中选择删除。

5. 在是否删除数据集？对话框中，在文本字段中输入 delete，然后点击删除。

6. 重复此过程中的步骤以删除您创建的第二个数据集：My_integration_dataset_final。

删除 Pub/Sub 订阅和主题

1. 在 Google Cloud 控制台中，转到 Pub/Sub 的订阅页面。

   转到“订阅”页面

2. 点击要删除的订阅旁边的复选框。在本教程中，请点击 my_integration_notifs_sub 订阅旁边的复选框。

3. 点击删除。

4. 在删除订阅对话框中，点击删除。

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

   转到“主题”页面

6. 点击 my_integration_notifs 主题旁边的复选框。

7. 点击删除。

8. 在删除主题对话框的文本字段中输入 delete，然后点击删除。

删除您的 Cloud Storage 存储桶

1. 在 Google Cloud 控制台中，前往 Cloud Storage 的浏览器页面。

   转到“浏览器”页面

2. 选中存储桶旁边的复选框。在本教程中，这是 my-integration-bucket。

3. 点击删除。

4. 在对话框的文本字段中输入 Delete，然后点击 Delete。

后续步骤

• 详细了解 Datastream。
• 使用旧版流式处理 API 将数据流式插入 BigQuery 以执行高级功能。
• 试用其他 Google Cloud 功能。查阅我们的教程。