并行运行 BigQuery 作业和 Workflows

1. 简介

1c05e3d0c2bd2b45 74be7b376d45258a

Workflows 是一种全代管式编排服务，可按照您定义的顺序执行 Google Cloud 或外部服务。

BigQuery 是一个全代管式企业数据仓库，具有机器学习、地理空间分析和商业智能等内置功能，可帮助您管理和分析 TB 级的数据。

在此 Codelab 中，您将针对公共维基百科数据集运行一些 BigQuery 查询。然后，您将了解如何作为 Workflows 编排的一部分以串行方式依次运行多个 BigQuery 查询。最后，您将使用 Workflows 的并行迭代功能并行执行查询，从而将速度提高多达 5 倍。

学习内容

  • 如何对维基百科数据集运行 BigQuery 查询。
  • 如何将多个查询作为 Workflows 编排的一部分依次运行。
  • 如何使用 Workflows 并行迭代并行执行查询，从而实现高达 5 倍的速度提升。

2. 设置和要求

自定进度的环境设置

  1. 登录 Google Cloud 控制台，然后创建一个新项目或重复使用现有项目。如果您还没有 Gmail 或 Google Workspace 账号，则必须创建一个。

b35bf95b8bf3d5d8.png

a99b7ace416376c4.png

bd84a6d3004737c5.png

  • 项目名称是此项目参与者的显示名称。它是 Google API 尚未使用的字符串。您可以随时对其进行更新。
  • 项目 ID 在所有 Google Cloud 项目中必须是唯一的，并且不可变（一经设置便无法更改）。Cloud 控制台会自动生成一个唯一字符串；通常您不在乎这是什么在大多数 Codelab 中，您都需要引用项目 ID（它通常标识为 PROJECT_ID）。如果您不喜欢生成的 ID，可以再随机生成一个 ID。或者，您也可以尝试自己的项目 ID，看看是否可用。完成此步骤后便无法更改该 ID，并且该 ID 在项目期间会一直保留。
  • 此外，还有第三个值，即某些 API 使用的项目编号，供您参考。如需详细了解所有这三个值，请参阅文档。
  1. 接下来，您需要在 Cloud 控制台中启用结算功能，以便使用 Cloud 资源/API。运行此 Codelab 应该不会产生太多的费用（如果有费用的话）。如需关停资源，以免产生超出本教程范围的结算费用，您可以删除自己创建的资源或删除整个项目。Google Cloud 的新用户符合参与 $300 USD 免费试用计划的条件。

启动 Cloud Shell

虽然可以通过笔记本电脑对 Google Cloud 进行远程操作，但在此 Codelab 中，您将使用 Google Cloud Shell，这是一个在云端运行的命令行环境。

在 Google Cloud 控制台 中，点击右上角工具栏中的 Cloud Shell 图标：

55efc1aaa7a4d3ad.png

预配和连接到环境应该只需要片刻时间。完成后，您应该会看到如下内容：

7ffe5cbb04455448.png

这个虚拟机已加载了您需要的所有开发工具。它提供了一个持久的 5 GB 主目录，并且在 Google Cloud 中运行，大大增强了网络性能和身份验证功能。您在此 Codelab 中的所有工作都可以在浏览器中完成。您无需安装任何程序。

3. 探索维基百科数据集

首先，浏览 BigQuery 中的维基百科数据集。

转到 Google Cloud 控制台的 BigQuery 部分：

ea75ab12a7c012a4.png

在 bigquery-samples 下，您应该会看到各种公共数据集，包括一些与维基百科相关的数据集：

c9484e305b8e1438.png

在 wikipedia_pageviews 数据集下，您可以看到不同年份的网页浏览量表格：

c540a4162640cbb3.png

您可以选择其中一个表（例如201207）并预览数据：

b5b2a334cd6f63c0.png

您还可以对表运行查询。例如，以下查询会选择观看次数最多的前 100 本图书：

SELECT TITLE, SUM(views)
FROM bigquery-samples.wikipedia_pageviews.201207h
GROUP BY TITLE
ORDER BY SUM(VIEWS) DESC
LIMIT 100

运行查询后，大约需要 20 秒来加载数据：

1df3877aed1653b4

4. 定义工作流以运行多个查询

对单个表运行查询非常简单。但是，针对多个表运行多个查询并整理结果可能会非常繁琐。Workflows 可以借助其迭代语法来协助实现这一点！

在 Cloud Shell 中，创建一个 workflow-serial.yaml 文件，以构建工作流来对多个表运行多个查询：

touch workflow-serial.yaml

然后，您可以在 Cloud Shell 中使用编辑器修改该文件：

33bf9325b078ad8

在 workflow-serial.yaml 文件的第一个 init 步骤中，创建一个 results 映射，以跟踪按表名称键控的每次迭代。另外，定义一个 tables 数组，其中包含您要对其运行查询的表的列表。在本例中，我们选择 5 个表：

main:
    steps:
    - init:
        assign:
            - results : {}
            - tables:
                - 201201h
                - 201202h
                - 201203h
                - 201204h
                - 201205h

接下来，定义 runQueries 步骤。此步骤将遍历每个表，并使用 Workflows 的使用 BigQuery 连接器运行查询，以查找每个表中网页浏览量最多的前 100 个图书。然后，它会在结果映射中保存每个表中的热门标题和视图：

    - runQueries:
        for:
            value: table
            in: ${tables}
            steps:
            - runQuery:
                call: googleapis.bigquery.v2.jobs.query
                args:
                    projectId: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
                    body:
                        useLegacySql: false
                        useQueryCache: false
                        timeoutMs: 30000
                        # Find the top 100 titles with most views on Wikipedia
                        query: ${
                            "SELECT TITLE, SUM(views)
                            FROM `bigquery-samples.wikipedia_pageviews." + table + "`
                            WHERE LENGTH(TITLE) > 10
                            GROUP BY TITLE
                            ORDER BY SUM(VIEWS) DESC
                            LIMIT 100"
                            }
                result: queryResult
            - returnResult:
                assign:
                    # Return the top title from each table
                    - results[table]: {}
                    - results[table].title: ${queryResult.rows[0].f[0].v}
                    - results[table].views: ${queryResult.rows[0].f[1].v}

在最后一步中，返回 results 映射：

    - returnResults:
        return: ${results}

5. 使用 Workflows 运行多个查询

在部署和运行工作流之前，您需要确保已启用 Workflows API。您可以通过 Google Cloud 控制台或在 Cloud Shell 中使用 gcloud 启用该 API：

gcloud services enable workflows.googleapis.com

为 Workflows 创建服务账号：

SERVICE_ACCOUNT=workflows-bigquery-sa
gcloud iam service-accounts create $SERVICE_ACCOUNT \
  --display-name="Workflows BigQuery service account"

请确保服务账号具有记录和运行 BigQuery 作业的角色：

PROJECT_ID=your-project-id
gcloud projects add-iam-policy-binding $PROJECT_ID \
  --role roles/logging.logWriter \
  --role roles/bigquery.jobUser \
  --member serviceAccount:$SERVICE_ACCOUNT@$PROJECT_ID.iam.gserviceaccount.com

使用服务账号部署工作流：

gcloud workflows deploy bigquery-serial \
    --source=workflow-serial.yaml \
    --service-account=$SERVICE_ACCOUNT@$PROJECT_ID.iam.gserviceaccount.com

最后，您就可以运行工作流了。

在 Cloud 控制台的“工作流”部分下，找到 bigquery-serial 工作流，然后点击 Execute 按钮：

b6afa4747680334f.png

或者，您也可以在 Cloud Shell 中使用 gcloud 运行工作流：

gcloud workflows run bigquery-serial

您应该会看到工作流执行会持续约 1 分钟（5 个表各 20 秒）。

最后，您会看到每个包含热门标题和视图的表格的输出：

304d11a5bffdada4.png

baf31533d3671c9e.png

6. 使用并行步骤并行处理多个查询

上一步中的工作流大约需要 1 分钟，因为它运行了 5 次查询，每次 20 秒。由于这些查询是独立的，因此实际上您可以使用 Workflows 的并行迭代功能并行运行这些查询。

将 workflow-serial.yaml 文件复制到新的 workflow-parallel.yaml 文件中。在新文件中，您将进行一些更改，以将串行步骤转换为并行步骤。

在 workflow-parallel.yaml 文件中，更改 runQueries 步骤。首先，添加 parallel 关键字。这样，for 循环的每次迭代都可以并行运行。然后，将 results 变量声明为 shared 变量。这样，分支就可以写入该变量。我们会将每个结果附加到此变量。

- runQueries:
    parallel:
        shared: [results]
        for:
            value: table
            in: ${tables}

部署并行工作流：

gcloud workflows deploy bigquery-parallel \
    --source=workflow-parallel.yaml \
    --service-account=$SERVICE_ACCOUNT@$PROJECT_ID.iam.gserviceaccount.com

运行工作流：

gcloud workflows run bigquery-parallel

您应该会看到工作流执行时长约为 20 秒。这是由于所有 5 个查询都是并行运行的。只需更改几行代码，速度提升高达 5 倍！

最后，您会看到每个表的输出都相同，它们的标题和视图都相同，但执行时间要短得多：

1825d49ef225c828

7. 恭喜

恭喜，您已完成此 Codelab！如需了解详情，请参阅有关并行步骤的工作流文档。

所学内容

  • 如何对维基百科数据集运行 BigQuery 查询。
  • 如何将多个查询作为 Workflows 编排的一部分依次运行。
  • 如何使用 Workflows 并行迭代并行执行查询，从而实现高达 5 倍的速度提升。