GSP290

概览

Dataflow 是一项 Google Cloud 服务，支持大规模的统一流式数据处理和批量数据处理操作。它基于 Apache Beam 项目构建，该项目是一种开源模型，用于定义批量数据和流式数据并行处理流水线。借助其中一个开源 Apache Beam SDK，您可以构建一个程序来定义流水线，然后使用 Dataflow 执行该流水线。

在本实验中，您将使用 Python 版 Apache Beam SDK 在 Dataflow 中构建和运行流水线，将 Cloud Storage 中的数据注入 BigQuery，然后在 BigQuery 中转换和丰富数据。

注意：请务必打开 Python 文件，并按照说明查看注释。这样可以了解代码的作用。

您将执行的任务

在本实验中，您将学习如何构建和运行 Dataflow 流水线 (Python)，以便执行以下任务：

• 将 Cloud Storage 中的数据注入 BigQuery。
• 在 BigQuery 中转换和丰富数据。
• 在 BigQuery 中联接数据并将结果写入新表。

设置和要求

点击“开始实验”按钮前的注意事项

请阅读以下说明。实验是计时的，并且您无法暂停实验。计时器在您点击开始实验后即开始计时，显示 Google Cloud 资源可供您使用多长时间。

此实操实验可让您在真实的云环境中开展实验活动，免受模拟或演示环境的局限。为此，我们会向您提供新的临时凭据，您可以在该实验的规定时间内通过此凭据登录和访问 Google Cloud。

为完成此实验，您需要：

• 能够使用标准的互联网浏览器（建议使用 Chrome 浏览器）。

注意：请使用无痕模式（推荐）或无痕浏览器窗口运行此实验。这可以避免您的个人账号与学生账号之间发生冲突，这种冲突可能导致您的个人账号产生额外费用。

• 完成实验的时间 - 请注意，实验开始后无法暂停。

注意：请仅使用学生账号完成本实验。如果您使用其他 Google Cloud 账号，则可能会向该账号收取费用。

如何开始实验并登录 Google Cloud 控制台

1. 点击开始实验按钮。如果该实验需要付费，系统会打开一个对话框供您选择支付方式。右侧是实验设置和访问权限面板，其中包含以下内容：

   • 打开 Google Cloud 控制台按钮
   • 您在本实验中必须使用的临时凭证（用户名和密码）
   • 帮助您逐步完成该实验所需的其他信息（如果需要）

   请注意，实验计时器位于页面顶部附近，将显示剩余时间。

2. 点击打开 Google Cloud 控制台（如果您使用的是 Chrome 浏览器，请右键点击并选择在无痕式窗口中打开链接）。

   该实验会启动资源并打开另一个标签页，显示“登录”页面。

   提示：可以将这些标签页分别放在不同的窗口中，并排显示。

   注意：如果您看见选择账号对话框，请点击使用其他账号。

3. 如有必要，请复制下方的用户名，然后将其粘贴到登录对话框中。

   {{{user_0.username | "<用户名>"}}}

   您也可以在实验设置和访问权限面板中找到“用户名”。

4. 点击下一步。

5. 复制下面的密码，然后将其粘贴到欢迎对话框中。

   {{{user_0.password | "<密码>"}}}

   您也可以在实验设置和访问权限面板中找到“密码”。

6. 点击下一步。

   重要提示：您必须使用实验提供的凭证。请勿使用您的 Google Cloud 账号凭证。 注意：在本实验中使用您自己的 Google Cloud 账号可能会产生额外费用。

7. 依次点击进入后续页面：

   • 接受条款及条件。
   • 由于这是临时账号，请勿添加账号恢复选项或双重身份验证。
   • 请勿注册免费试用。

片刻之后，系统会在此标签页中打开 Google Cloud 控制台。

注意：如需访问 Google Cloud 产品和服务，请点击导航菜单，或在搜索字段中输入服务或产品的名称。

激活 Cloud Shell

Cloud Shell 是一种装有开发者工具的虚拟机。它提供了一个永久性的 5GB 主目录，并且在 Google Cloud 上运行。Cloud Shell 提供可用于访问您的 Google Cloud 资源的命令行工具。

1. 点击 Google Cloud 控制台顶部的激活 Cloud Shell 。

2. 在弹出的窗口中执行以下操作：

   • 继续完成 Cloud Shell 信息窗口中的设置。
   • 授权 Cloud Shell 使用您的凭据进行 Google Cloud API 调用。

如果您连接成功，即表示您已通过身份验证，且项目 ID 会被设为您的 Project_ID 。输出内容中有一行说明了此会话的 Project_ID：

Your Cloud Platform project in this session is set to {{{project_0.project_id | "PROJECT_ID"}}}

gcloud 是 Google Cloud 的命令行工具。它已预先安装在 Cloud Shell 上，且支持 Tab 自动补全功能。

3. （可选）您可以通过此命令列出活跃账号名称：

gcloud auth list

4. 点击授权。

输出：

ACTIVE: * ACCOUNT: {{{user_0.username | "ACCOUNT"}}} To set the active account, run: $ gcloud config set account `ACCOUNT`

5. （可选）您可以通过此命令列出项目 ID：

gcloud config list project

输出：

[core] project = {{{project_0.project_id | "PROJECT_ID"}}} 注意：如需查看在 Google Cloud 中使用 gcloud 的完整文档，请参阅 gcloud CLI 概览指南。

任务 1. 确保 Dataflow API 已成功启用

• 为了确保能访问这个必要的 API，请重新启动与 Dataflow API 的连接。

  重要提示：即使该 API 目前已启用，也请按照以下步骤停用并重新启用该 API，以便成功将其重新启动。

gcloud services disable dataflow.googleapis.com --project {{{project_0.project_id|Project ID}}} --force gcloud services enable dataflow.googleapis.com --project {{{project_0.project_id|Project ID}}}

重新启用该 API 后，页面会显示停用选项。

点击检查我的进度以验证您已完成的任务。 停用并重新启用 Dataflow API。

任务 2. 下载起始代码

下载要在本实验中使用的 Dataflow Python 示例。

• 在 Cloud Shell 中运行以下命令，从 Google Cloud 的专业服务 GitHub 中获取 Dataflow Python 示例：

gcloud storage cp -r gs://spls/gsp290/dataflow-python-examples .

任务 3. 创建 Cloud Storage 存储桶并将文件复制到该存储桶

在 Cloud Shell 中创建一个 Cloud Storage 存储桶，然后将文件复制到该存储桶。这些文件是 Dataflow Python 示例。

创建 Cloud Storage 存储桶

• 在 Cloud Shell 中，使用 make bucket 命令在项目的 区域中新建一个区域级存储桶：

gcloud storage buckets create gs://{{{ project_0.project_id | BUCKET_NAME }}} --location={{{ project_0.default_region | REGION }}}

点击检查我的进度以验证您已完成的任务。 创建 Cloud Storage 存储桶。

将文件复制到存储桶

• 在 Cloud Shell 中，使用 gsutil 命令将文件复制到刚才创建的 Cloud Storage 存储桶中：

gcloud storage cp gs://spls/gsp290/data_files/usa_names.csv gs://{{{ project_0.project_id | BUCKET_NAME }}}/data_files/ gcloud storage cp gs://spls/gsp290/data_files/head_usa_names.csv gs://{{{ project_0.project_id | BUCKET_NAME }}}/data_files/

点击检查我的进度以验证您已完成的任务。 将文件复制到存储桶。

任务 4. 创建 BigQuery 数据集

创建一个 BigQuery 数据集。您的表将加载到 BigQuery 中的这个数据集中。

• 在 Cloud Shell 中，创建名为 lake 的数据集：

bq mk lake

点击检查我的进度以验证您已完成的任务。 创建名为“lake”的 BigQuery 数据集。

任务 5. 查看和运行数据注入流水线

在此任务中，您将查看流水线代码，了解其工作原理。然后，您将设置并运行流水线。

数据注入流水线使用 TextIO 作为来源、BigQueryIO 作为目的地，将 Cloud Storage 中的数据注入到 BigQuery 表中。具体而言，该流水线将执行以下操作：

• 从 Cloud Storage 提取文件。
• 滤除文件中的标题行。
• 将读取的行转换为字典对象。
• 将行输出到 BigQuery。

在 Cloud Shell IDE 中启用 Gemini Code Assist

您可以在集成开发环境 (IDE)（例如 Cloud Shell）中使用 Gemini Code Assist，以获取代码方面的指导或解决代码问题。您需要先启用 Gemini Code Assist，然后才能使用该功能。

1. 在 Cloud Shell 中，使用以下命令启用 Gemini for Google Cloud API：

gcloud services enable cloudaicompanion.googleapis.com

2. 点击 Cloud Shell 工具栏上的打开编辑器。

注意：如需打开 Cloud Shell 编辑器，请点击 Cloud Shell 工具栏上的打开编辑器。您可以根据需要点击打开编辑器或打开终端，在 Cloud Shell 和代码编辑器之间切换。

3. 点击屏幕底部状态栏中的 Cloud Code - 无项目。

4. 按照说明对插件进行授权。如果系统未自动选择项目，请点击选择 Google Cloud 项目，然后选择 。

5. 检查您的 Google Cloud 项目 () 是否显示在状态栏的 Cloud Code 状态消息中。

查看数据注入流水线的 Python 代码

在本部分中，请向 Gemini Code Assist 发出提示，以便获取有关数据注入流水线的更多信息，从而为新团队成员提供概览。

1. 在 Cloud Shell 编辑器的文件探索器中，前往 dataflow_python_examples > dataflow_python_examples > data_ingestion.py。

2. 打开 data_ingestion.py 文件。此操作会启用 Gemini Code Assist，启用后，编辑器右上角会显示 图标。

3. 点击 Gemini Code Assist：智能操作 图标，然后选择 Explain this（解释此内容）。

4. Gemini Code Assist 会打开一个聊天窗格，其中预填充了 Explain this 提示。在 Code Assist 聊天窗格的内嵌文本框中，将预填充的提示替换为以下内容，然后点击发送：

You are an expert Data Engineer at Cymbal AI. A new team member is unfamiliar with this pipeline code. Explain the purpose and functionality of the data ingestion pipeline defined in the data_ingestion.py. Your explanation should include: 1. A high-level summary of what the script does. 2. A breakdown of the key components, such as the DataIngestion class and the run function. 3. An explanation of how the script uses the Apache Beam pipeline to read, process, and write data. 4. The role of command-line arguments and how they are used. 5. A description of the input data format and the output BigQuery table schema. For the suggested improvements, don't update this file.

这些代码会使用 Cloud Storage 中的数据文件填充 BigQuery 表。data_ingestion.py 文件中代码的详细说明会显示在 Gemini Code Assist 聊天中。

5. 如需返回 Cloud Shell，请点击打开终端。

为 Dataflow 作业设置 Docker 容器

在本部分中，您将返回 Cloud Shell 会话，以便设置所需的 Python 库。

在本实验中，Dataflow 作业需要使用 Python3.8。为确保您使用的是正确的版本，请在 Python 3.8 Docker 容器中运行 Dataflow 进程。

1. 在 Cloud Shell 中运行以下命令来启动 Python 容器：

cd ~ docker run -it -e PROJECT={{{ project_0.project_id | PROJECT_ID }}} -v $(pwd)/dataflow-python-examples:/dataflow python:3.8 /bin/bash

此命令会拉取包含 Python 3.8 最新稳定版的 Docker 容器，并执行一个命令 shell，以便在容器内运行后续命令。-v 标志会将源代码作为容器的一个卷提供，这样，我们在 Cloud Shell 编辑器中修改源代码的同时，仍能在运行的容器中访问源代码。

2. 容器完成拉取并开始在 Cloud Shell 中执行后，运行以下命令将 apache-beam 安装在运行的容器中：

pip install apache-beam[gcp]==2.59.0

3. 接下来，在 Cloud Shell 中运行的容器内，将目录切换到链接源代码的位置：

cd dataflow/

在云端运行数据注入流水线

1. 运行以下代码以执行数据注入流水线：

python dataflow_python_examples/data_ingestion.py \ --project={{{ project_0.project_id | PROJECT_ID }}} \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --temp_location gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --input gs://{{{ project_0.project_id | BUCKET_NAME }}}/data_files/head_usa_names.csv \ --save_main_session

这些代码会启动所需的工作器，并在流水线完成后将它们关闭。

2. 在控制台标题栏的搜索字段中，输入 Dataflow，然后点击搜索结果中的 Dataflow。

Dataflow 页面打开后，查看作业的状态。

3. 点击作业的名称以查看其进度。

作业状态显示为成功后，您便可执行下一步。此注入流水线从启动、完成工作到关闭，大约需要五分钟时间。

4. 前往 BigQuery（导航菜单 > BigQuery），确认是否已填充数据。

5. 点击项目名称，查看 lake 数据集下的 usa_names 表。

6. 点击此表，然后前往预览标签页，查看 usa_names 数据的示例。

注意：如果您没有看到 usa_names 表，请尝试刷新页面或使用经典版 BigQuery 界面来查看此表。

点击检查我的进度以验证您已完成的任务。 构建数据注入流水线。

任务 6. 查看和运行数据转换流水线

在此任务中，您将查看数据转换流水线，了解其工作原理。然后，您将运行该流水线来处理 Cloud Storage 文件，并将结果输出到 BigQuery。

数据转换流水线还将使用 TextIO 作为来源、BigQueryIO 作为目的地，将 Cloud Storage 中的数据注入 BigQuery 表中，但会进行额外的数据转换。具体而言，该流水线将执行以下操作：

• 从 Cloud Storage 提取文件。
• 将读取的行转换为字典对象。
• 将包含年份的数据转换为 BigQuery 视为日期的格式。
• 将行输出到 BigQuery。

查看数据转换流水线的 Python 代码

在本部分中，您将输入提示，让 Gemini Code Assist 提供有关数据转换流水线的更多信息，以便进一步为新团队成员提供帮助。

1. 在 Cloud Shell 菜单栏中点击打开编辑器。

2. 在 Cloud Shell 编辑器中的同一目录下，找到 data_transformation.py 文件。与之前一样，请注意编辑器右上角的 图标。

3. 点击 Gemini Code Assist：智能操作 图标，然后选择 Explain this（解释此内容）。

4. Gemini Code Assist 会打开一个聊天窗格，其中预填充了 Explain this 提示。在 Code Assist 聊天窗格的内嵌文本框中，将预填充的提示替换为以下内容，然后点击发送：

You are an expert Data Engineer at Cymbal AI. A new team member is unfamiliar with this pipeline code. Explain the purpose and functionality of the data transformation pipeline defined in the data_transformation.py. Your explanation should include: 1. A high-level summary of what the script does, noting its differences from a simple ingestion pipeline. 2. A breakdown of the key components, specifically the DataTransformation class and the run function. 3. A detailed explanation of how the script uses the Apache Beam pipeline to read from a file, transform the data, and write it to a BigQuery table. 4. Describe how the script handles the BigQuery schema by reading it from a JSON file. 5. Explain the data transformation logic within the parse_method, particularly how it converts the year to a DATE type. 6. The role of command-line arguments and how they are used. For the suggested improvements, don't update this file.

针对 data_transformation.py 文件中代码的说明会显示在 Gemini Code Assist 聊天中。

在云端运行数据转换流水线

1. 在 Cloud Shell 终端中输入以下命令，以便运行数据转换流水线：

python dataflow_python_examples/data_transformation.py \ --project={{{ project_0.project_id | PROJECT_ID }}} \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --temp_location gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --input gs://{{{ project_0.project_id | BUCKET_NAME }}}/data_files/head_usa_names.csv \ --save_main_session

2. 在 Google Cloud 控制台标题栏的搜索字段中输入 Dataflow，然后点击搜索结果中的 Dataflow。

3. 点击此作业名称可查看作业状态。

此 Dataflow 流水线从启动、完成工作到关闭，大约需要五分钟时间。

4. 当 Dataflow“作业状态”屏幕中的作业状态显示为成功后，前往 BigQuery 确认是否已填充数据。

您应该可在 lake 数据集下看到 usa_names_transformed 表。

5. 点击此表，然后前往预览标签页，查看 usa_names_transformed 数据的示例。

注意：如果您没有看到 usa_names_transformed 表，请尝试刷新页面或使用经典版 BigQuery 界面来查看此表。

点击检查我的进度以验证您已完成的任务。 构建数据转换流水线。

任务 7. 查看并运行数据丰富化流水线

现在，您将构建一个数据丰富化流水线，以实现以下目标：

• 从 Cloud Storage 提取文件。
• 滤除文件中的标题行。
• 将读取的行转换为字典对象。
• 将行输出到 BigQuery。

查看和修改数据丰富化流水线的 Python 代码

在本部分中，您将利用 Gemini Code Assist 中 AI 赋能的功能，来查看和修改数据丰富化流水线的 Python 代码。

1. 在 Cloud Shell 菜单栏中点击打开编辑器。

2. 在 Cloud Shell 编辑器中，仍在同一目录中，找到 data_enrichment.py。与之前一样，请注意编辑器右上角的 图标。

3. 点击工具栏上的 Gemini Code Assist：智能操作 图标。

4. 如需更新第 83 行的代码，请将以下提示粘贴到从工具栏中打开的 Gemini Code Assist 内嵌文本字段中。

In the data_enrichment.py file, update line 83 by replacing x.decode('utf8') with x.

5. 如需提示 Gemini Code Assist 按要求修改代码，请按 Enter 键。

6. 在 Gemini Diff 视图中看到提示时，点击接受。

data_enrichment.py 文件中更新后的第 83 行现在如下所示：

values = [x for x in csv_row]

7. 将该行代码修改完毕后，请务必在代码编辑器中选择文件选项并点击保存，以保存更新后的文件。

运行数据丰富化流水线

1. 在 Cloud Shell 终端中输入以下命令，以便执行数据丰富化流水线：

python dataflow_python_examples/data_enrichment.py \ --project={{{ project_0.project_id | PROJECT_ID }}} \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --temp_location gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --input gs://{{{ project_0.project_id | BUCKET_NAME }}}/data_files/head_usa_names.csv \ --save_main_session

2. 在 Dataflow 页面中，点击您的作业以查看作业状态。

此 Dataflow 流水线从启动、完成工作到关闭，大约需要五分钟时间。

3. 当 Dataflow 作业状态屏幕中的作业状态显示为成功后，在控制台中依次点击导航菜单 > BigQuery，确认是否已填充数据。

您应该可在 lake 数据集下看到 usa_names_enriched 表。

4. 点击此表，然后前往预览标签页，查看 usa_names_enriched 数据的示例。

注意：如果您没有看到 usa_names_enriched 表，请尝试刷新页面或使用经典版 BigQuery 界面来查看此表。

点击检查我的进度以验证您已完成的任务。 构建数据丰富化流水线。

任务 8. 查看和运行从数据湖到数据集市的流水线

接下来，您需要构建一条 Dataflow 流水线，从两个 BigQuery 数据源读取数据，然后联接这两个数据源。具体操作如下：

• 从两个 BigQuery 数据源提取文件。
• 联接两个数据源。
• 滤除文件中的标题行。
• 将读取的行转换为字典对象。
• 将行输出到 BigQuery。

运行数据注入流水线以执行数据联接，并将生成的表写入 BigQuery

您首先要查看 data_lake_to_mart.py 代码，了解其作用。然后，您将在云端运行该流水线。

1. 在代码编辑器中，打开 data_lake_to_mart.py 文件。

查看文件中的注释，这些注释解释了代码的作用。此代码会联接两个表，并将结果写入 BigQuery 中的新表。

注意：如需详细了解从数据湖到数据集市的流水线，您可以像在任务 5 和任务 6 中那样，再次运用 Gemini Code Assist 中 AI 赋能的功能来解释代码。

2. 运行以下代码块以执行流水线：

python dataflow_python_examples/data_lake_to_mart.py \ --worker_disk_type="compute.googleapis.com/projects//zones//diskTypes/pd-ssd" \ --max_num_workers=4 \ --project={{{ project_0.project_id | PROJECT_ID }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --temp_location gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --save_main_session \ --region={{{ project_0.default_region | REGION }}}

3. 在 Google Cloud 控制台标题栏的搜索字段中输入 Dataflow，然后点击搜索结果中的 Dataflow。

4. 点击此新作业以查看作业状态。

此 Dataflow 流水线从启动、完成工作到关闭，大约需要五分钟时间。

5. 当 Dataflow“作业状态”屏幕中的作业状态显示为成功后，点击导航菜单 > BigQuery 确认是否已填充数据。

您应该可在 lake 数据集下看到 orders_denormalized_sideinput 表。

6. 点击此表，然后前往预览部分，查看 orders_denormalized_sideinput 数据的示例。

注意：如果您没有看到 orders_denormalized_sideinput 表，请尝试刷新页面或使用经典版 BigQuery 界面来查看此表。

点击检查我的进度以验证您已完成的任务。 构建从数据湖到数据集市的 Dataflow 流水线

检验您的掌握情况

下面有一道选择题可强化您对此实验所涉概念的理解。请尽您所能回答。

恭喜！

您借助 Gemini Code Assist，使用 Dataflow 执行了 Python 代码，将 Cloud Storage 中的数据注入到 BigQuery 中，然后在 BigQuery 中转换并丰富了数据。

后续步骤/了解详情

想要查找更多参考信息？请点击以下链接查看官方文档：

• Dataflow
• BigQuery
• 查看 Apache Beam 程序设计指南，了解高级概念。
• 探索以下实验：
  • 在 BQML 中使用分类模型预测访问者购买情况
  • Cloud Managed Service for Airflow：跨不同位置复制 BigQuery 表

Google Cloud 培训和认证

…可帮助您充分利用 Google Cloud 技术。我们的课程会讲解各项技能与最佳实践，可帮助您迅速上手使用并继续学习更深入的知识。我们提供从基础到高级的全方位培训，并有点播、直播和虚拟三种方式选择，让您可以按照自己的日程安排学习时间。各项认证可以帮助您核实并证明您在 Google Cloud 技术方面的技能与专业知识。

本手册的最后更新时间：2025 年 9 月 1 日

本实验的最后测试时间：2025 年 9 月 1 日

版权所有 2026 Google LLC 保留所有权利。Google 和 Google 徽标是 Google LLC 的商标。其他所有公司名和产品名可能是其各自相关公司的商标。