写入 Airflow DAG

Cloud Composer 1 | Cloud Composer 2

本指南介绍如何编写在 Cloud Composer 环境中运行的 Apache Airflow 有向无环图 (DAG)。

由于 Apache Airflow 不提供强大的 DAG 和任务隔离,我们建议您使用单独的生产环境和测试环境,以防止 DAG 干扰。如需了解详情,请参阅测试 DAG

构建 Airflow DAG

Airflow DAG 在 Python 文件中定义,由以下组件组成:

  • DAG 定义
  • Airflow 运算符
  • 运算符关系

以下代码段脱离上下文显示了每个组成部分的示例。

DAG 定义

以下示例演示了 DAG 定义:

import datetime

from airflow import models

default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    "start_date": datetime.datetime(2018, 1, 1),
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    "composer_sample_simple_greeting",
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:

运算符和任务

运算符用于描述要完成的工作。任务任务是操作的特定实例。

from airflow.operators import bash_operator
from airflow.operators import python_operator

    def greeting():
        import logging

        logging.info("Hello World!")

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = python_operator.PythonOperator(
        task_id="hello", python_callable=greeting
    )

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = bash_operator.BashOperator(
        task_id="bye", bash_command="echo Goodbye."
    )

任务关系

任务关系描述了工作的完成顺序。

# Define the order in which the tasks complete by using the >> and <<
# operators. In this example, hello_python executes before goodbye_bash.
hello_python >> goodbye_bash

Python 中的完整 DAG 工作流示例

以下工作流是一个完整的有效 DAG 模板,由 hello_python 任务和 goodbye_bash 任务组成:


import datetime

from airflow import models

from airflow.operators import bash_operator
from airflow.operators import python_operator

default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    "start_date": datetime.datetime(2018, 1, 1),
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    "composer_sample_simple_greeting",
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:
    def greeting():
        import logging

        logging.info("Hello World!")

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = python_operator.PythonOperator(
        task_id="hello", python_callable=greeting
    )

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = bash_operator.BashOperator(
        task_id="bye", bash_command="echo Goodbye."
    )

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

如需详细了解如何定义 Airflow DAG,请参阅 Airflow 教程Airflow 概念

Airflow 运算符

以下示例演示了几个常用的 Airflow 运算符。如需查看 Airflow 运算符的权威参考信息,请参阅运算符和钩子参考文档以及提供程序索引

BashOperator

BashOperator 用于运行命令行程序。

from airflow.operators import bash

    # Create BigQuery output dataset.
    make_bq_dataset = bash.BashOperator(
        task_id="make_bq_dataset",
        # Executing 'bq' command requires Google Cloud SDK which comes
        # preinstalled in Cloud Composer.
        bash_command=f"bq ls {bq_dataset_name} || bq mk {bq_dataset_name}",
    )

Cloud Composer 在 Airflow 工作器上的 Bash 脚本中运行提供的命令。工作器是一个基于 Debian 的 Docker 容器,其中包含多个软件包。

PythonOperator

PythonOperator 用于运行任意 Python 代码。

Cloud Composer 在一个容器中运行 Python 代码,该容器包含您的环境中所用 Cloud Composer 映像版本的软件包。

如需安装其他 Python 软件包,请参阅安装 Python 依赖项

Google Cloud 运算符

如需运行使用 Google Cloud 产品的任务,请使用 Google Cloud Airflow 运算符。例如,BigQuery 运算符BigQuery 中查询和处理数据。

还有许多适用于 Google Cloud 的 Airflow 运算符以及 Google Cloud 提供的单项服务。如需查看完整列表,请参阅 Google Cloud 运维人员

from airflow.providers.google.cloud.operators import bigquery
from airflow.providers.google.cloud.transfers import bigquery_to_gcs

    bq_recent_questions_query = bigquery.BigQueryInsertJobOperator(
        task_id="bq_recent_questions_query",
        configuration={
            "query": {
                "query": RECENT_QUESTIONS_QUERY,
                "useLegacySql": False,
                "destinationTable": {
                    "projectId": project_id,
                    "datasetId": bq_dataset_name,
                    "tableId": bq_recent_questions_table_id,
                },
            }
        },
        location=location,
    )

EmailOperator

使用 EmailOperator 从 DAG 发送电子邮件。如需从 Cloud Composer 环境发送电子邮件,您必须将您的环境配置为使用 SendGrid

from airflow.operators import email

    # Send email confirmation (you will need to set up the email operator
    # See https://cloud.google.com/composer/docs/how-to/managing/creating#notification
    # for more info on configuring the email operator in Cloud Composer)
    email_summary = email.EmailOperator(
        task_id="email_summary",
        to="{{var.value.email}}",
        subject="Sample BigQuery notify data ready",
        html_content="""
        Analyzed Stack Overflow posts data from {min_date} 12AM to {max_date}
        12AM. The most popular question was '{question_title}' with
        {view_count} views. Top 100 questions asked are now available at:
        {export_location}.
        """.format(
            min_date=min_query_date,
            max_date=max_query_date,
            question_title=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][0] }}"
            ),
            view_count=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][1] }}"
            ),
            export_location=output_file,
        ),
    )

运营商故障通知

如果您希望在 DAG 中的运算符发生失败时发送电子邮件通知,可以将 email_on_failure 设置为 True。如需从 Cloud Composer 环境发送电子邮件通知,您必须将您的环境配置为使用 SendGrid

from airflow import models

default_dag_args = {
    "start_date": yesterday,
    # Email whenever an Operator in the DAG fails.
    "email": "{{var.value.email}}",
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "project_id": project_id,
}

with models.DAG(
    "composer_sample_bq_notify",
    schedule_interval=datetime.timedelta(weeks=4),
    default_args=default_dag_args,
) as dag:

DAG 工作流准则

  1. 将任何自定义 Python 库放入嵌套目录下一个 DAG 的 ZIP 归档文件中。不要将这些库放入 DAG 目录顶层。

    当 Airflow 扫描 dags/ 文件夹时,Airflow 只会检查以下 Python 模块中的 DAG:位于 DAG 文件夹顶层的 Python 模块,以及位于 ZIP 归档文件顶层(位于顶层 dags/ 文件夹)中的 DAG。如果 Airflow 在某个 ZIP 归档文件中遇到既不包含 airflow 又不包含 DAG 子字符串的 Python 模块,则 Airflow 会停止处理该 ZIP 归档文件。Airflow 仅返回在此之前发现的 DAG。

  2. 为实现容错功能,请不要在同一个 Python 模块中定义多个 DAG 对象。

  3. 请勿使用 SubDAG。请改为在 DAG 内对任务进行分组

  4. 将 DAG 解析时所需的文件放入 dags/ 文件夹(而不是 data/ 文件夹中)。

  5. 为 DAG 实现单元测试

  6. 按照测试 DAG 的说明中的建议对开发或修改的 DAG 进行测试。

  7. Composer 本地开发 CLI 工具通过在本地运行 Airflow 环境来简化 Cloud Composer 2 的 Apache Airflow DAG 开发。此本地 Airflow 环境使用特定 Cloud Composer 2 版本的映像。

  8. 确认开发的 DAG 不会过多增加 DAG 解析时间

  9. Airflow 任务可能因多种原因而失败。为避免整个 DAG 运行失败,我们建议您启用任务重试功能。将重试次数上限设置为 0 表示不执行重试。

    我们建议使用任务停用值(而非 0替换 default_task_retries 选项。此外,您还可以在任务级别设置 retries 参数

  10. 如果要在 Airflow 任务中使用 GPU,请使用具有 GPU 的机器根据节点创建单独的 GKE 集群。使用 GKEStartPodOperator 运行任务。

  11. 避免在运行其他 Airflow 组件(调度器、工作器、Web 服务器)的集群的节点池中运行占用大量 CPU 和内存的任务。请改用 KubernetesPodOperatorGKEStartPodOperator

  12. 将 DAG 部署到环境中时,请仅将解读和执行 DAG 所必需的文件上传至 /dags 文件夹。

  13. 限制 /dags 文件夹中的 DAG 文件数量。

    Airflow 正在不断解析 /dags 文件夹中的 DAG。解析是一个循环遍历 DAG 文件夹的过程,需要加载的文件数(及其依赖项)会影响 DAG 解析和任务调度的性能。使用 100 个文件,每个包含 100 个 DAG,而不是使用 10000 个文件,每个文件包含 1 个 DAG,效率要高得多,因此建议进行此类优化。这种优化在解析时间与 DAG 编写和管理效率之间的平衡。

    您还可以考虑(例如)要部署 10000 个 DAG 文件,您可以创建 100 个 ZIP 文件,每个文件包含 100 个 DAG 文件。

    除了上述提示之外,如果您拥有超过 10000 个 DAG 文件,那么以编程方式生成 DAG 或许是一个不错的选择。例如,您可以实现一个 Python DAG 文件,以生成一定数量的 DAG 对象(例如 200100 个 DAG 对象)。

有关编写 DAG 的常见问题解答

如果我想在多个 DAG 中运行相同或类似的任务,如何尽量减少重复代码?

我们建议定义库和封装容器,以最大限度地减少重复代码。

如何在 DAG 文件之间重复使用代码?

将您的实用函数放入一个本地 Python 库中并导入这些函数。您可以在环境存储桶内 dags/ 文件夹中的任何 DAG 中引用这些函数。

如何尽量降低出现不同定义的风险?

例如,您有两个团队希望将原始数据汇总成收入指标。这两个团队分别编写了两项略有不同的任务,完成相同的任务。建议针对收入数据来定义库,这样,DAG 实现者就必须阐明要汇总的收入的定义。

如何设置 DAG 之间的依赖项?

这取决于您想要如何定义依赖项。

如果您有两个 DAG(即 DAG A 和 DAG B),并且希望 DAG B 在 DAG A 之后触发,则可以在 DAG A 末尾添加 TriggerDagRunOperator

如果 DAG B 仅依赖于 DAG A 生成的工件(例如 Pub/Sub 消息),那么可能更适合使用传感器。

如果 DAG B 与 DAG A 紧密集成,则或许可以将两个 DAG 合并为一个 DAG。

如何将唯一运行 ID 传递给某一 DAG 及其任务?

例如,您想传递 Dataproc 集群名称和文件路径。

您可以通过在 PythonOperator 中返回 str(uuid.uuid4()) 来随机生成一个唯一 ID。这会将 ID 放入 XComs 中,以便您通过模板化字段在其他运算符中引用该 ID。

在生成 uuid 之前,请考虑 DagRun 专用的 ID 是否更加有用。您还可以使用在 Jinja 替代变量中引用这些 ID。

如何在 DAG 中分离任务?

每项任务都应该是一个具有幂等性的工作单元。因此,应避免将一个涉及多个步骤的工作流封装到单项任务中,例如,在 PythonOperator 中运行一个复杂程序。

如果我需要汇总多个来源中的数据,那么是否应该在一个 DAG 中定义多项任务?

例如,您有多个包含原始数据的表格,并且希望针对每个表格创建每日汇总数据。这些任务并不相互依赖。在这种情况下,您是应该为每个表格分别创建一项任务和一个 DAG,还是应该创建一个通用 DAG?

如果您能接受各项任务共用相同的 DAG 级属性(例如 schedule_interval),那么最好在一个 DAG 中定义多项任务。否则,可以通过一个 Python 模块生成多个 DAG(将这些 DAG 放入该模块的 globals() 中即可),以尽量减少重复代码。

如何限制在一个 DAG 中运行的并发任务数量?

例如,您想避免超出 API 用量限额或配额,或避免同时运行过多进程。

您可以在 Airflow 网页界面中定义 Airflow 池,并将任务与 DAG 中的现有池相关联。

有关使用运算符的常见问题解答

我是否应该使用 DockerOperator

我们不建议使用 DockerOperator,除非它用于在安装的远程 Docker 上(而不是在环境的集群中)启动容器。在 Cloud Composer 环境中,操作员无权访问 Docker 守护程序。

请改用 KubernetesPodOperatorGKEStartPodOperator。这些运算符分别将 Kubernetes pod 启动到 Kubernetes 或 GKE 集群中。请注意,我们不建议将 Pod 启动到环境的集群中,因为这可能会导致资源竞争。

我是否应该使用 SubDagOperator

我们不建议您使用 SubDagOperator

请按照对 Tasks 进行分组说明中提供的替代方案。

如果我想将 Python 运算符完全隔离,是否应该仅在 PythonOperators 中运行 Python 代码?

有几种方案可供您选择,具体取决于您的目标。

如果您只想维护单独的 Python 依赖项,可以使用 PythonVirtualenvOperator

考虑使用 KubernetesPodOperator。您可以使用此运算符定义 Kubernetes Pod 并在其他集群中运行这些 Pod。

如何添加自定义二进制文件或非 PyPI 软件包?

您可以安装托管在私有软件包代码库中的软件包

您还可以使用 KubernetesPodOperator 通过自己用自定义软件包构建的映像来运行 Kubernetes pod。

如何将参数统一传递给某一 DAG 及其任务?

您可以使用 Airflow 对 Jinja 模板化的内置支持传递可在模板化字段中使用的参数。

何时会发生模板替换?

在开始调用运算符的 pre_execute 函数之前,系统会在 Airflow 工作器上进行模板替换。实际上,这意味着模板只有在任务即将运行之前才会被替换。

如何确定哪些运算符参数支持模板替换?

支持 Jinja2 模板替换的运算符参数带有明确标记。

在运算符定义中查找 template_fields 字段,该字段包含接受模板替换的参数名称列表。

例如,请参阅 BashOperator,它支持 bash_commandenv 参数模板化。

后续步骤