Cloud Composer 1 处于维护后模式。Google 无法发布 Cloud Composer 1 的进一步更新，包括新版 Airflow、问题修复和安全更新。我们建议您规划迁移到 Cloud Composer 2。

此页面由 Cloud Translation API 翻译。

写入 Airflow DAG

Cloud Composer 1 |Cloud Composer 2 |Cloud Composer 3

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

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

构建 Airflow DAG

Airflow DAG 在 Python 文件中定义，由以下几个组成部分：

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

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

DAG 定义

以下示例演示了 Airflow 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:

运算符和任务

Airflow 操作器描述了要完成的工作。任务 task 是运算符的特定实例。

from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

    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 = PythonOperator(task_id="hello", python_callable=greeting)

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = 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.bash import BashOperator
from airflow.operators.python import PythonOperator



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 = PythonOperator(task_id="hello", python_callable=greeting)

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = 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 工作器。工作器是一个基于 Debian 的 Docker 容器，其中包含多个软件包

gcloud 命令，包括用于处理 Cloud Storage 存储桶的 gcloud storage 子命令。
bq 命令
kubectl 命令

PythonOperator

使用 PythonOperator 运行任意 Python 代码。

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

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

Google Cloud 运算符

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

还有许多适用于 Google Cloud 和 Google Cloud 提供的各项服务的 Airflow 运算符。请参阅如需查看完整列表，请参阅 Google Cloud Operators。

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 工作流指南

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

Airflow 在扫描 dags/ 文件夹时，只会检查以下 Python 模块中的 DAG：位于 DAGs 文件夹顶层的 Python 模块，以及位于某一 ZIP 归档文件（该文件也位于顶层 dags/ 文件夹）顶层的 Python 模块。如果 Airflow 在 ZIP 归档文件中遇到了 Python 模块，该模块不包含 airflow 和 DAG 子字符串，Airflow 会停止处理该 ZIP 文件归档。并仅返回在此之前发现的 DAG。
为实现容错功能，请不要在同一个 Python 模块中定义多个 DAG 对象。
请勿使用 SubDAG。而是将 DAG 中的任务分组。
将 DAG 解析时所需的文件放入 dags/ 文件夹，而不是位于 data/ 文件夹中。
为 DAG 实现单元测试。
按照测试 DAG 的说明。
Composer Local Development CLI 工具简化了通过运行 Airflow 为 Cloud Composer 2 开发 Apache Airflow DAG 环境此本地 Airflow 环境使用特定 Cloud Composer 2 版本的映像。
确认开发的 DAG 不会过多增加 DAG 解析时间。
Airflow 任务可能会因多种原因而失败。为避免整个 DAG 运行，我们建议您启用任务重试功能。将重试次数上限设置为 0 表示不执行任何重试。

我们建议替换 default_task_retries选项，并将值设置为 0 以外的任务重试次数。此外，您还可以在任务级别设置 retries 参数。
如果您想在 Airflow 任务中使用 GPU，请根据使用带有 GPU 的机器的节点创建单独的 GKE 集群。使用 GKEStartPodOperator 运行任务。
避免在运行其他 Airflow 组件（调度器、工作器、Web 服务器）的集群节点池中运行 CPU 和内存密集型任务。请改用 KubernetesPodOperator 或 GKEStartPodOperator。
将 DAG 部署到环境时，请仅将解析和执行 DAG 所绝对必要的文件上传到 /dags 文件夹。
限制 /dags 文件夹中的 DAG 文件数量。

Airflow 会持续解析 /dags 文件夹中的 DAG。解析是循环遍历 DAG 文件夹的过程，需要加载的文件数量（及其依赖项）会影响 DAG 解析和任务调度的性能。使用 100 个文件（每个文件包含 100 个 DAG）比使用 10,000 个文件（每个文件包含 1 个 DAG）的效率要高得多，因此建议进行此类优化。此优化在解析时间与 DAG 编写和管理效率之间取得了平衡。

例如，您还可以考虑部署 10,000 个 DAG 文件创建 100 个 ZIP 文件，每个文件包含 100 个 DAG 文件。

除了上述提示之外，如果您有超过 10,000 个 DAG 文件，则通过编程方式生成 DAG 可能是一个不错的选择。例如，您可以实现一个 Python DAG 文件，用于生成一定数量的 DAG 对象（例如 20、100 个 DAG 对象）。

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

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

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

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

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

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

例如，您有两个团队希望将原始数据汇总成收入指标。为此，这两个团队都各自编写了一个略微不同的任务。建议针对收入数据来定义库，这样，DAG 实现者就必须阐明要汇总的收入的定义。

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

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

如果您有两个 DAG（DAG A 和 DAG B），并且希望 DAG B 在 DAG 之后触发 A，您可以在 TriggerDagRunOperator：DAG A 末尾。

如果 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（除非将其用于启动容器（而不是在环境的）。在 Cloud Composer 环境中，操作员没有对 Docker 守护程序的访问权限。

请改用 KubernetesPodOperator 或 GKEStartPodOperator。这些运算符可将 Kubernetes Pod Kubernetes 或 GKE 集群。请注意，建议将 Pod 启动到环境的集群中，因为这可能会导致再到资源竞争

我是否应该使用 `SubDagOperator`？

我们不建议您使用 SubDagOperator。

按照任务分组中的建议，使用替代方法。

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

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

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

请考虑使用 KubernetesPodOperator。使用此运算符定义 Kubernetes Pod 并在其他集群中运行 Pod

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

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

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

您可以使用 Airflow 的内置支持 Jinja 模板，用于传递可以使用的参数。

何时会发生模板替换？

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

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

明确支持 Jinja2 模板替换的运算符参数如此处所示。

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

有关示例，请参见 BashOperator，它支持 bash_command 和 env 参数。