在受管理的 Apache Spark 集群上运行 Hadoop 字数统计作业

1. 简介

工作流是数据分析中的常见用例，涉及提取、转换和分析数据，以找出其中的有意义信息。在 Google Cloud Platform 中，用于编排工作流的工具是 Managed Apache Airflow，它是热门开源工作流工具 Apache Airflow 的托管版本。在本实验中，您将使用 Managed Apache Airflow 创建一个简单的工作流，该工作流会创建 Managed Apache Spark 集群，使用 Managed Apache Spark 和 Apache Hadoop 对其进行分析，然后删除 Managed Apache Spark 集群。

什么是 Managed Apache Airflow？

Managed Apache Airflow 是一项全托管式工作流编排服务，您可以利用它编写、安排和监管跨越多项云服务和本地数据中心的流水线。Managed Apache Airflow 在备受欢迎的 Apache Airflow 开源项目基础上构建而成，通过 Python 编程语言操作，易于使用且没有供应商锁定。

通过使用 Managed Apache Airflow（而不是 Apache Airflow 的本地实例），用户可以从 Airflow 的强大功能中获益，而不会产生安装或管理开销。

如需详细了解 Managed Apache Airflow，请参阅 文档！

什么是 Apache Airflow？

Apache Airflow 是一款开源工具，用于以编程方式编写、安排和监控工作流。您需要记住一些与 Airflow 相关的重要术语，这些术语将在整个实验中出现：

• DAG - DAG（有向无环图）是您要安排和运行的有序任务的集合。DAG 也称为工作流，在标准 Python 文件中定义
• 运算符 - 运算符描述工作流中的单个任务

什么是 Managed Apache Spark？

Managed Apache Spark 是 Google Cloud Platform 的全托管式 Apache Spark 和 Apache Hadoop 服务。Managed Apache Spark 可轻松与其他 GCP 服务集成，为您提供一个强大而完善的平台来执行数据处理、分析和机器学习等任务。

如需详细了解 Managed Apache Spark，请参阅 文档！

您将执行 的操作

本 Codelab 将向您介绍如何在 Managed Apache Airflow 中创建并运行 Apache Airflow 工作流，该工作流将完成以下任务：

• 创建 Managed Apache Spark 集群
• 在该集群上运行 Apache Hadoop WordCount 作业，并将其结果输出到 Cloud Storage
• 删除集群

您将学习的内容

• 如何在 Managed Apache Airflow 中创建并运行 Apache Airflow 工作流
• 如何使用 Managed Apache Airflow 和 Managed Apache Spark 对数据集运行分析
• 如何通过 Google Cloud Platform 控制台、Cloud SDK 和 Airflow 网页界面访问 Managed Apache Airflow 环境

所需条件

• GCP 账号
• 基本的 CLI 知识
• 对 Python 有基本的了解

2. 设置 GCP

创建项目

选择或创建 Google Cloud Platform 项目。

记下您的项目 ID，以便在后续步骤中使用。

如果您要创建新项目，项目 ID 位于创建页面上的“项目名称”下方
如果您已创建项目，则可以在控制台首页的“项目信息”卡片中找到该 ID

注意：如果您不打算保留在本教程中创建的资源，请创建一个新项目，而不是选择现有项目。完成本教程后，您可以删除所创建的项目，从而移除与该项目和教程关联的所有资源。

启用 API

启用 Managed Apache Airflow、Managed Apache Spark 和 Cloud Storage API。启用后，您可以忽略显示“前往凭据”的按钮，然后继续执行本教程的下一步。

如需详细了解如何启用和停用 API，请参阅 文档

创建 Managed Apache Airflow 环境

使用以下配置创建 Managed Apache Airflow 环境： 名称：my-airflow-environment 位置：us-central1 可用区：us-central1-a 所有其他配置都可以保留默认值。点击底部的“创建”。

当 GCP 控制台中的环境 页面上环境名称左侧显示绿色对勾标记时，即表示环境创建过程已完成。创建过程需要几分钟时间，您可以在环境创建过程中离开此页面。

创建 Cloud Storage 存储分区

在您的项目中，创建 Cloud Storage 存储分区，配置如下： 名称：<your-project-id> 默认存储类别：多区域 位置：美国 访问权限控制模型：精细 准备就绪后，按“创建”

3. 设置 Apache Airflow

如果您的 Managed Apache Airflow 环境尚未创建完成，请跳到“示例工作流”步骤，稍后再返回此部分

查看 Managed Apache Airflow 环境信息

在 GCP 控制台中，打开“环境”页面

点击环境名称以查看其详细信息。

环境详情 页面会提供 Airflow 网页界面网址、Google Kubernetes Engine 集群 ID、Cloud Storage 存储分区的名称以及 /dags 文件夹的路径等信息。

在 Airflow 中，DAG（有向无环图）是您要安排和运行的有序任务的集合。DAG 也称为工作流，在标准 Python 文件中定义。Managed Apache Airflow 只会安排 /dags 文件夹中的 DAG。/dags 文件夹位于 Managed Apache Airflow 在您创建环境时自动创建的 Cloud Storage 存储分区中。

设置 Apache Airflow 环境变量

Apache Airflow 变量是 Airflow 特有的一个概念，此类变量不同于 环境变量。在此步骤中，您将设置以下三个 Airflow 变量：gcp_project、gcs_bucket 和 gce_zone。

使用 gcloud 设置变量

首先，打开您的 Cloud Shell，其中已为您方便地安装了 Cloud SDK。

将环境变量 COMPOSER_INSTANCE 设置为 Managed Apache Airflow 环境的名称

COMPOSER_INSTANCE=my-airflow-environment

如需使用 gcloud 命令行工具设置 Airflow 变量，请使用带有 variables 子命令的 gcloud composer environments run 命令。此 gcloud composer 命令会执行 Airflow CLI 子命令 variables。该子命令会将参数传递给 gcloud 命令行工具。

您将运行此命令三次，并将变量替换为与您的项目相关的变量。

使用以下命令设置 gcp_project，并将 <your-project-id> 替换为您在第 2 步中记下的项目 ID。

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --set gcp_project <your-project-id>

输出将如下所示

kubeconfig entry generated for us-central1-my-airflow-env-123abc-gke.
Executing within the following Kubernetes cluster namespace: composer-1-10-0-airflow-1-10-2-123abc
[2020-04-17 20:42:49,713] {settings.py:176} INFO - settings.configure_orm(): Using pool settings. pool_size=5, pool_recycle=1800, pid=449
[2020-04-17 20:42:50,123] {default_celery.py:90} WARNING - You have configured a result_backend of redis://airflow-redis-service.default.svc.cluste
r.local:6379/0, it is highly recommended to use an alternative result_backend (i.e. a database).
[2020-04-17 20:42:50,127] {__init__.py:51} INFO - Using executor CeleryExecutor
[2020-04-17 20:42:50,433] {app.py:52} WARNING - Using default Composer Environment Variables. Overrides have not been applied.
[2020-04-17 20:42:50,440] {configuration.py:522} INFO - Reading the config from /etc/airflow/airflow.cfg
[2020-04-17 20:42:50,452] {configuration.py:522} INFO - Reading the config from /etc/airflow/airflow.cfg

使用以下命令设置 gcs_bucket，并将 <your-bucket-name> 替换为您在第 2 步中记下的存储分区 ID。如果您遵循了我们的建议，则存储分区名称与项目 ID 相同。您的输出将与上一个命令类似。

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --set gcs_bucket gs://<your-bucket-name>

使用以下命令设置 gce_zone。您的输出将与之前的命令类似。

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --set gce_zone us-central1-a

（可选）使用 gcloud 查看变量

如需查看变量的值，请运行带有 get 实参的 Airflow CLI 子命令 variables，或使用 Airflow 界面。

例如：

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --get gcs_bucket

您可以对刚刚设置的三个变量中的任意一个执行此操作：gcp_project、gcs_bucket 和 gce_zone。

4. 示例工作流

我们来看一下将在第 5 步中使用的 DAG 的代码。暂时不用担心下载文件，只需按照此处的说明操作即可。

这里有很多内容需要解压缩，我们来稍微分解一下。

from airflow import models
from airflow.contrib.operators import dataproc_operator
from airflow.utils import trigger_rule

我们首先导入一些 Airflow：

• airflow.models - 允许我们在 Airflow 数据库中访问和创建数据。
• airflow.contrib.operators - 社区中的运算符所在的位置。在本例中，我们需要 dataproc_operator 来访问 Managed Apache Spark API。
• airflow.utils.trigger_rule - 用于向运算符添加触发规则。触发规则允许对运算符是否应相对于其父项的状态执行进行精细控制。

output_file = os.path.join(
    models.Variable.get('gcs_bucket'), 'wordcount',
    datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep

这指定了输出文件的位置。这里值得注意的一行是 models.Variable.get('gcs_bucket')，它将从 Airflow 数据库中获取 gcs_bucket 变量值。

WORDCOUNT_JAR = (
    'file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar'
)

input_file = 'gs://pub/shakespeare/rose.txt'

wordcount_args = ['wordcount', input_file, output_file]

• WORDCOUNT_JAR - 我们最终将在 Managed Apache Spark 集群上运行的 .jar 文件的位置。它已为您托管在 GCP 上。
• input_file - 包含 Hadoop 作业最终将计算的数据的文件所在的位置。我们将在第 5 步中一起将数据上传到该位置。
• wordcount_args - 我们将传递到 jar 文件的实参。

yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1),
    datetime.datetime.min.time())

这将为我们提供一个相当于前一天午夜的 datetime 对象。例如，如果此操作在 3 月 4 日上午 11:00 执行，则 datetime 对象将表示 3 月 3 日 00:00。这与 Airflow 处理调度的方式有关。如需了解详情，请点击此处。

default_dag_args = {
    'start_date': yesterday,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': models.Variable.get('gcp_project')
}

每当创建新的 DAG 时，都应以字典的形式提供 default_dag_args 变量：

• 'email_on_failure' - 指示任务失败时是否应发送电子邮件提醒
• 'email_on_retry' - 指示任务重试时是否应发送电子邮件提醒
• 'retries' - 表示在 DAG 失败的情况下，Airflow 应尝试重试多少次
• 'retry_delay' - 表示 Airflow 在尝试重试之前应等待多长时间
• 'project_id' - 告知 DAG 要将其与哪个 GCP 项目 ID 相关联，稍后将需要使用 Managed Apache Spark 运算符

with models.DAG(
        'composer_hadoop_tutorial',
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:

使用 with models.DAG 会告知脚本将下面的所有内容都包含在同一个 DAG 中。我们还会看到传入了三个实参：

• 第一个实参是一个字符串，用于指定要创建的 DAG 的名称。在本例中，我们使用的是 composer_hadoop_tutorial。
• schedule_interval - 一个 datetime.timedelta 对象，我们在此处将其设置为一天。这意味着，此 DAG 将在之前在 'default_dag_args' 中设置的 'start_date' 之后每天尝试执行一次
• default_args - 我们之前创建的字典，其中包含 DAG 的默认实参

创建 Managed Apache Spark 集群

接下来，我们将创建一个 dataproc_operator.DataprocClusterCreateOperator，用于创建 Managed Apache Spark 集群。

    create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(
        task_id='create_dataproc_cluster',
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        num_workers=2,
        zone=models.Variable.get('gce_zone'),
        master_machine_type='n1-standard-1',
        worker_machine_type='n1-standard-1')

在此运算符中，我们会看到几个实参，除了第一个实参之外，所有实参都是此运算符特有的：

• task_id - 与 BashOperator 中一样，这是我们分配给运算符的名称，可从 Airflow 界面中查看
• cluster_name - 我们分配给 Managed Apache Spark 集群的名称。在这里，我们将其命名为 composer-hadoop-tutorial-cluster-{{ ds_nodash }}（如需了解可选的其他信息，请参阅信息框）
• num_workers - 我们分配给 Managed Apache Spark 集群的工作器数量
• zone - 我们希望集群所在的地理区域，保存在 Airflow 数据库中。这将读取我们在第 3 步中设置的 'gce_zone' 变量
• master_machine_type - 我们希望分配给 Managed Apache Spark 主节点的机器类型
• worker_machine_type - 我们希望分配给 Managed Apache Spark 工作器的机器类型

有关 cluster_name 变量的其他信息

参数的 {{ ds_nodash }} 部分之所以存在，是因为 Airflow 支持 jinja2 模板。这是一个参数，Airflow 会在运行时每次启动运算符时呈现该参数。在本例中，{{ ds_nodash }} 将替换为 DAG 的 execution_date，格式为 YYYYMMDD。如果您不熟悉 jinja2 模板，也没关系！完成本实验不需要任何相关知识。

提交 Apache Hadoop 作业

dataproc_operator.DataProcHadoopOperator 允许我们向 Managed Apache Spark 集群提交作业。

    run_dataproc_hadoop = dataproc_operator.DataProcHadoopOperator(
        task_id='run_dataproc_hadoop',
        main_jar=WORDCOUNT_JAR,
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        arguments=wordcount_args)

我们提供了几个参数：

• task_id - 我们分配给 DAG 的这一部分的名称
• main_jar - 我们希望针对集群运行的 .jar 文件的位置
• cluster_name - 要针对其运行作业的集群的名称，您会注意到它与我们在上一个运算符中找到的名称相同
• arguments - 传递到 jar 文件的实参，就像从命令行执行 .jar 文件时一样

删除集群

我们将创建的最后一个运算符是 dataproc_operator.DataprocClusterDeleteOperator。

    delete_dataproc_cluster = dataproc_operator.DataprocClusterDeleteOperator(
        task_id='delete_dataproc_cluster',
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        trigger_rule=trigger_rule.TriggerRule.ALL_DONE)

顾名思义，此运算符将删除给定的 Managed Apache Spark 集群。我们在此处看到三个实参：

• task_id - 与 BashOperator 中一样，这是我们分配给运算符的名称，可从 Airflow 界面中查看
• cluster_name - 我们分配给 Managed Apache Spark 集群的名称。在这里，我们将其命名为 composer-hadoop-tutorial-cluster-{{ ds_nodash }}（如需了解可选的其他信息，请参阅“创建 Managed Apache Spark 集群”后的信息框）
• trigger_rule - 我们在本步骤开始时的导入过程中简要提及了触发规则，但在这里我们有一个正在使用的触发规则。默认情况下，除非所有上游运算符都已成功完成，否则 Airflow 运算符不会执行。ALL_DONE 触发规则仅要求所有上游运算符都已完成，无论它们是否成功。在这里，这意味着即使 Hadoop 作业失败，我们仍然希望关闭集群。

  create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

最后，我们希望这些运算符按特定顺序执行，并且可以使用 Python 位移运算符来表示这一点。在本例中，create_dataproc_cluster 将始终首先执行，然后是 run_dataproc_hadoop，最后是 delete_dataproc_cluster。

综上，代码如下所示：

# Copyright 2018 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# [START composer_hadoop_tutorial]
"""Example Airflow DAG that creates a Managed Apache Spark cluster, runs the Hadoop
wordcount example, and deletes the cluster.

This DAG relies on three Airflow variables
https://airflow.apache.org/concepts.html#variables
* gcp_project - Google Cloud Project to use for the Managed Apache Spark cluster.
* gce_zone - Google Compute Engine zone where Managed Apache Spark cluster should be
  created.
* gcs_bucket - Google Cloud Storage bucket to use for result of Hadoop job.
  See https://cloud.google.com/storage/docs/creating-buckets for creating a
  bucket.
"""

import datetime
import os

from airflow import models
from airflow.contrib.operators import dataproc_operator
from airflow.utils import trigger_rule

# Output file for Managed Apache Spark job.
output_file = os.path.join(
    models.Variable.get('gcs_bucket'), 'wordcount',
    datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep
# Path to Hadoop wordcount example available on every Managed Apache Spark cluster.
WORDCOUNT_JAR = (
    'file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar'
)
# Arguments to pass to Managed Apache Spark job.
input_file = 'gs://pub/shakespeare/rose.txt'

wordcount_args = ['wordcount', input_file, output_file]

yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1),
    datetime.datetime.min.time())

default_dag_args = {
    # Setting start date as yesterday starts the DAG immediately when it is
    # detected in the Cloud Storage bucket.
    'start_date': yesterday,
    # To email on failure or retry set 'email' arg to your email and enable
    # emailing here.
    'email_on_failure': False,
    'email_on_retry': False,
    # If a task fails, retry it once after waiting at least 5 minutes
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': models.Variable.get('gcp_project')
}

# [START composer_hadoop_schedule]
with models.DAG(
        'composer_hadoop_tutorial',
        # Continue to run DAG once per day
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:
    # [END composer_hadoop_schedule]

    # Create a Managed Apache Spark cluster.
    create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(
        task_id='create_dataproc_cluster',
        # Give the cluster a unique name by appending the date scheduled.
        # See https://airflow.apache.org/code.html#default-variables
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        num_workers=2,
        zone=models.Variable.get('gce_zone'),
        master_machine_type='n1-standard-1',
        worker_machine_type='n1-standard-1')

    # Run the Hadoop wordcount example installed on the Managed Apache Spark cluster
    # master node.
    run_dataproc_hadoop = dataproc_operator.DataProcHadoopOperator(
        task_id='run_dataproc_hadoop',
        main_jar=WORDCOUNT_JAR,
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        arguments=wordcount_args)

    # Delete Managed Apache Spark cluster.
    delete_dataproc_cluster = dataproc_operator.DataprocClusterDeleteOperator(
        task_id='delete_dataproc_cluster',
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        # Setting trigger_rule to ALL_DONE causes the cluster to be deleted
        # even if the Managed Apache Spark job fails.
        trigger_rule=trigger_rule.TriggerRule.ALL_DONE)

    # [START composer_hadoop_steps]
    # Define DAG dependencies.
    create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster
    # [END composer_hadoop_steps]

# [END composer_hadoop]

如果您因环境未创建而跳过了第 3 步，请务必立即返回该步骤，然后再继续执行第 5 步。

5. 将 Airflow 文件上传到 Cloud Storage

将 DAG 复制到 /dags 文件夹

1. 首先，打开您的 Cloud Shell，其中已为您方便地安装了 Cloud SDK。
2. 克隆 Python 示例代码库并切换到 composer/workflows 目录

git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git && cd python-docs-samples/composer/workflows

3. 运行以下命令，将 DAG 文件夹的名称设置为环境变量

DAGS_FOLDER=$(gcloud composer environments describe ${COMPOSER_INSTANCE} \
--location us-central1 --format="value(config.dagGcsPrefix)")

4. 运行以下 gsutil 命令，将教程代码复制到创建 /dags 文件夹的位置

gsutil cp hadoop_tutorial.py $DAGS_FOLDER

输出将如下所示：

Copying file://hadoop_tutorial.py [Content-Type=text/x-python]...
/ [1 files][  4.1 KiB/  4.1 KiB]
Operation completed over 1 objects/4.1 KiB.

注意：本指南现在将工作流称为 composer_hadoop_tutorial

6. 使用 Airflow 界面

如需使用 GCP 控制台访问 Airflow 网页界面，请执行以下操作： 打开环境 页面。 在环境对应的 Airflow Web 服务器 列中，点击新窗口图标。Airflow 网页界面会在新的浏览器窗口中打开。

如需了解 Airflow 界面，请参阅 访问网页界面。

查看变量

您之前设置的变量会保留在您的环境中。您可以从 Airflow 界面菜单栏中选择 Admin > Variables 来查看变量。

探索 DAG 运行

当您将 DAG 文件上传到 Cloud Storage 中的 dags 文件夹时，Cloud Composer 会解析该文件。如果未发现任何错误，工作流的名称会显示在 DAG 列表中，并且工作流会排入队列以立即运行。如需查看 DAG，请点击页面顶部的 DAG 。

点击 composer_hadoop_tutorial 以打开 DAG 详情页面。此页面包含工作流任务和依赖项的图形表示。

现在，在工具栏中点击 Graph View （图表视图），然后将鼠标悬停在每个任务对应的图形化表示上，查看其状态。请注意，每个任务的边框也指示状态（绿色边框表示任务正在运行中；红色边框表示任务失败，等等）。

如需从 Graph View 重新运行工作流，请执行以下操作：

1. 在 Airflow 界面的“Graph View”中，点击 create_dataproc_cluster 图形。
2. 点击 Clear （清除）以重置三个任务，然后点击 OK （确定）进行确认。

您还可以转到以下 GCP 控制台页面来检查 composer-hadoop-tutorial 工作流的状态和结果：

• Managed Apache Spark 集群：可以监控集群的创建和删除。请注意，由工作流创建的集群是临时性的，也就是说，此类集群仅在工作流的持续期间内存在，并且将在最后一个工作流任务的执行过程中删除。
• Managed Apache Spark 作业：可以查看或监控 Apache Hadoop WordCount 作业。点击“任务 ID”即可查看作业日志输出。
• Cloud Storage 浏览器：可以在您为本 Codelab 创建的 Cloud Storage 存储分区所含的 wordcount 文件夹中查看 WordCount 的结果。

7. 清理

为避免系统因本 Codelab 中使用的资源向您的 GCP 账号收取费用，请执行以下操作：

1. （可选）如需保存数据，请从 Managed Apache Airflow 环境的 Cloud Storage 存储分区以及您为本 Codelab 创建的存储分区中下载数据。
2. 删除您为本 Codelab 创建的 Cloud Storage 存储分区。
3. 删除环境的 Cloud Storage 存储分区。
4. 删除 Managed Apache Airflow 环境。请注意，删除环境并不会删除环境的存储分区。

您还可以选择删除项目：

1. 在 GCP 控制台中，转到 **项目** 页面。
2. 在项目列表中，选择要删除的项目，然后点击删除 。
3. 在框中输入项目 ID，然后点击关停 以删除项目。

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

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