Airflow 系列6：DAG 代码编写与 Operator 参数详解

airflow

发布日期: 2023-07-05

文章字数: 5.9k

阅读时长: 24 分

阅读次数:

上一篇文章中已经可以编写一个简单 DAG，并且在 Airflow Web 页面中看到 spark_layer_demo 的运行状态。

但是如果只是把示例代码跑起来，其实还不够。

真正写 Airflow 任务时，经常会遇到下面这些问题：

default_args 应该放什么参数？
DAG(...) 中的 schedule_interval、start_date、catchup 到底怎么理解？
BashOperator 除了 bash_command 还有哪些常用参数？
什么时候用 BashOperator，什么时候用 PythonOperator？
任务失败后怎么重试？
多个任务之间的依赖怎么写更清楚？
Airflow 中常见的 Operator 类型有哪些？

所以这篇文章继续基于 Apache Airflow 2.3.4，专门整理 DAG 代码的编写方式和 Operator 常用参数。

一、先看完整 DAG 示例

先看当前使用的 spark_layer_demo.py：

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.bash import BashOperator


default_args = {
    "owner": "test_owner",
    "depends_on_past": False,
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}


with DAG(
    dag_id="spark_layer_demo",
    default_args=default_args,
    description="Airflow DAG dependency demo: dwd -> dws -> ads",
    schedule_interval=timedelta(days=1),
    start_date=datetime(2023, 6, 21),
    catchup=False,
    tags=["spark", "demo"],
) as dag:

    dwd = BashOperator(
        task_id="dwd",
        bash_command="echo 'submit dwd spark job'; date",
        retries=3,
    )

    dws = BashOperator(
        task_id="dws",
        bash_command="echo 'submit dws spark job'; date",
        retries=3,
    )

    ads = BashOperator(
        task_id="ads",
        bash_command="echo 'submit ads spark job'; date",
        retries=3,
    )

    dwd >> dws >> ads

这段代码可以拆成 5 个部分：

代码位置	作用
`import`	导入 DAG、Operator、时间工具
`default_args`	定义任务默认参数
`with DAG(...) as dag`	定义一个 DAG
`BashOperator(...)`	定义具体任务
`dwd >> dws >> ads`	定义任务依赖关系

Airflow 解析这个文件时，不是简单地从上到下执行一个脚本，而是通过 Python 代码构造出一个 DAG 对象和多个 Task 对象。

Scheduler 解析完成后，会把 DAG 信息、Task 信息、运行状态写入元数据库。

DAG 文件解析成 DAG 和 Task 的关系图

这张图中需要重点理解的是：DAG 文件本质是 Python 代码，Airflow Scheduler 解析文件后，会得到 DAG 对象、Task 对象和任务依赖关系。

真正运行时，DAG 的定义信息、DAG Run 的状态、Task Instance 的状态，都会继续写入元数据库中。

二、导包方式

Airflow 2.x 中推荐使用新的导包路径。

例如 BashOperator：

from airflow.operators.bash import BashOperator

PythonOperator：

from airflow.operators.python import PythonOperator

BranchPythonOperator：

from airflow.operators.python import BranchPythonOperator

EmptyOperator：

from airflow.operators.empty import EmptyOperator

有些旧文档中会看到下面这种写法：

from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator

在 Airflow 2.3.4 中，旧写法在一些场景下仍然可能兼容，但是新文章和新代码建议使用 Airflow 2.x 的导包方式。

这样后续升级 Airflow 版本时，代码会更清晰。

三、`default_args` 的作用

default_args 是一组默认参数。

它会作用到 DAG 中的 Operator 上。

例如：

default_args = {
    "owner": "test_owner",
    "depends_on_past": False,
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

这里的意思是：

默认任务负责人是 test_owner。
当前任务不依赖上一个调度周期的同名任务。
任务失败不发送邮件。
任务重试不发送邮件。
默认失败后重试 1 次。
每次重试间隔 5 分钟。

常见参数如下：

参数	示例	说明
`owner`	`"test_owner"`	任务负责人，Web UI 中会显示
`depends_on_past`	`False`	是否依赖上一个调度周期的同一个任务成功
`email`	`["xxx@qq.com"]`	邮件接收人
`email_on_failure`	`False`	失败时是否发送邮件
`email_on_retry`	`False`	重试时是否发送邮件
`retries`	`1`	失败后重试次数
`retry_delay`	`timedelta(minutes=5)`	每次重试间隔
`retry_exponential_backoff`	`False`	是否使用指数退避重试
`execution_timeout`	`timedelta(hours=1)`	单个任务最大执行时长
`start_date`	`datetime(2023, 6, 21)`	任务级别开始时间，一般放 DAG 上更清楚

需要注意：default_args 不是只能给 DAG 使用。

它真正影响的是 DAG 内的任务。

例如下面的代码：

default_args = {
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    dag_id="demo",
    default_args=default_args,
    start_date=datetime(2023, 6, 21),
    schedule_interval=None,
) as dag:
    t1 = BashOperator(
        task_id="t1",
        bash_command="echo t1",
    )

虽然 t1 没有单独写 retries，但是它会继承 default_args 中的 retries=1。

如果任务自己又写了一次参数，以任务自己的参数为准：

t1 = BashOperator(
    task_id="t1",
    bash_command="echo t1",
    retries=3,
)

这时 t1 的重试次数就是 3，不是 default_args 中的 1。

四、`DAG(...)` 常用参数

DAG(...) 用来定义一个工作流。

当前示例中：

with DAG(
    dag_id="spark_layer_demo",
    default_args=default_args,
    description="Airflow DAG dependency demo: dwd -> dws -> ads",
    schedule_interval=timedelta(days=1),
    start_date=datetime(2023, 6, 21),
    catchup=False,
    tags=["spark", "demo"],
) as dag:
    ...

常用参数说明如下：

参数	示例	说明
`dag_id`	`"spark_layer_demo"`	DAG 唯一标识
`default_args`	`default_args`	任务默认参数
`description`	`"xxx"`	DAG 描述信息
`schedule_interval`	`timedelta(days=1)`	调度周期
`start_date`	`datetime(2023, 6, 21)`	调度起始逻辑时间
`catchup`	`False`	是否补跑历史周期
`tags`	`["spark", "demo"]`	Web UI 中的标签
`max_active_runs`	`1`	同一个 DAG 最多同时运行多少个 DAG Run
`concurrency`	`16`	DAG 级别最大并发任务数
`dagrun_timeout`	`timedelta(hours=2)`	DAG Run 最大运行时间

4.1 `dag_id`

dag_id 是 DAG 的唯一标识。

后续很多地方都会用到它：

$ airflow dags list
$ airflow dags trigger spark_layer_demo
$ airflow tasks list spark_layer_demo

元数据库中也会使用这个字段。

例如：

SELECT dag_id, is_paused, next_dagrun
FROM dag
WHERE dag_id = 'spark_layer_demo';

所以 dag_id 不建议随便修改。

如果一个 DAG 已经运行过，后来修改了 dag_id，Airflow 会认为这是一个新的 DAG。

4.2 `schedule_interval`

schedule_interval 表示调度周期。

常见写法有几种。

第一种，使用 timedelta：

schedule_interval=timedelta(days=1)

第二种，使用 Cron 表达式：

schedule_interval="0 2 * * *"

表示每天凌晨 2 点执行一次。

第三种，使用预设值：

schedule_interval="@daily"

常见预设值如下：

写法	含义
`@once`	只运行一次
`@hourly`	每小时
`@daily`	每天
`@weekly`	每周
`@monthly`	每月
`@yearly`	每年

如果只想手动触发，不希望 Airflow 自动调度，可以写：

schedule_interval=None

4.3 `start_date`

start_date 表示调度的起始逻辑时间。

它不是代码创建时间，也不是文件放入 dags 目录的时间。

例如：

start_date=datetime(2023, 6, 21)

配合：

schedule_interval=timedelta(days=1)

表示 Airflow 会从 2023-06-21 这个逻辑时间开始计算调度周期。

这里有一个容易误解的点：

Airflow 页面中的 execution_date 或 Logical Date，不一定等于任务真实开始运行的时间。

调度任务通常会在一个数据周期结束后才被创建。

所以写 DAG 时不要把 start_date 理解成“马上运行”的时间。

4.4 `catchup`

catchup 用来控制是否补跑历史周期。

如果：

start_date=datetime(2023, 1, 1)
schedule_interval="@daily"
catchup=True

当前时间已经是 2023 年 7 月，那么 Airflow 可能会尝试创建大量历史 DAG Run。

学习和测试环境通常建议：

catchup=False

这样只关注最新一次调度。

生产环境是否开启 catchup，要看业务是否需要补算历史数据。

例如离线数仓任务，如果某几天数据漏跑，可能就需要补跑。

但是如果只是监控任务、接口调用任务，通常不希望补跑大量历史任务。

五、`BashOperator` 参数详解

BashOperator 用来执行 Bash 命令、脚本或一组 Shell 命令。

当前示例中：

dwd = BashOperator(
    task_id="dwd",
    bash_command="echo 'submit dwd spark job'; date",
    retries=3,
)

常用参数如下：

参数	示例	说明
`task_id`	`"dwd"`	当前任务的唯一标识
`bash_command`	`"echo hello"`	要执行的 Shell 命令
`env`	`{"ENV": "test"}`	传给命令的环境变量
`append_env`	`True`	是否继承当前环境变量并追加 `env`
`cwd`	`"/tmp"`	命令执行目录
`output_encoding`	`"utf-8"`	输出编码
`skip_exit_code`	`99`	指定退出码为 skipped 状态
`retries`	`3`	当前任务单独设置重试次数
`retry_delay`	`timedelta(minutes=5)`	当前任务单独设置重试间隔

5.1 `task_id`

task_id 是任务唯一标识。

同一个 DAG 中不能有两个相同的 task_id。

例如：

dwd = BashOperator(
    task_id="dwd",
    bash_command="echo dwd",
)

后面查看日志、查看任务状态、查询元数据库，都会用到 task_id。

$ airflow tasks logs spark_layer_demo dwd <run_id>

SELECT dag_id, task_id, run_id, state
FROM task_instance
WHERE dag_id = 'spark_layer_demo'
  AND task_id = 'dwd';

5.2 `bash_command`

bash_command 是真正要执行的命令。

可以是一条命令：

bash_command="date"

也可以是多条命令：

bash_command="echo start; date; echo end"

也可以写成多行字符串：

bash_command="""
echo "start dwd"
date
echo "end dwd"
"""

提交 Spark 任务时可以写成：

bash_command="""
/opt/module/spark-yarn/bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master yarn \
  /opt/module/spark-yarn/examples/jars/spark-examples_2.12-3.1.3.jar 10
"""

如果 Spark 不在 Airflow 机器上，而是在其他机器上，也可以通过 SSH 执行：

bash_command="""
ssh hadoop102 "/opt/module/spark-yarn/bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master yarn \
  /opt/module/spark-yarn/examples/jars/spark-examples_2.12-3.1.3.jar 10"
"""

5.3 建议加 `set -e`

写多条 Bash 命令时，建议在前面加：

set -e

例如：

bash_command="""
set -e
echo "start dwd"
python /opt/scripts/dwd.py
echo "end dwd"
"""

原因是：

如果中间某条子命令失败，但是整个 Shell 最终退出码仍然是 0，Airflow 可能会认为任务成功。

加上 set -e 后，只要中间命令失败，整个 Bash 任务就会失败。

这个对排查任务状态非常重要。

5.4 `env` 和模板变量

如果需要给 Bash 命令传递参数，可以使用 env。

例如：

dwd = BashOperator(
    task_id="dwd",
    bash_command='echo "biz_date=$BIZ_DATE"',
    env={
        "BIZ_DATE": "{{ ds }}",
    },
)

这里的 {{ ds }} 是 Airflow 的模板变量，表示当前任务的逻辑日期，格式通常是 YYYY-MM-DD。

这种写法比直接在 bash_command 中拼接复杂参数更清楚。

如果参数来自手动触发时传入的 dag_run.conf，也建议通过 env 传递：

dwd = BashOperator(
    task_id="dwd",
    bash_command='echo "message=$MESSAGE"',
    env={
        "MESSAGE": '{{ dag_run.conf["message"] if dag_run else "" }}',
    },
)

这样可以减少命令拼接中的转义问题。

5.5 `skip_exit_code`

BashOperator 默认把退出码 0 作为成功。

默认情况下，如果命令退出码是 99，任务会进入 skipped 状态。

例如：

check_file = BashOperator(
    task_id="check_file",
    bash_command="""
    if [ ! -f /tmp/data.txt ]; then
      exit 99
    fi
    echo "file exists"
    """,
)

如果 /tmp/data.txt 不存在，任务不会标记为 failed，而是 skipped。

如果希望所有非 0 退出码都认为失败，可以设置：

skip_exit_code=None

六、`PythonOperator` 参数详解

PythonOperator 用来执行一个 Python 函数。

示例：

from airflow.operators.python import PythonOperator


def print_context(ds, **kwargs):
    print(f"ds={ds}")
    print(kwargs)


task = PythonOperator(
    task_id="print_context",
    python_callable=print_context,
)

常用参数如下：

参数	示例	说明
`task_id`	`"print_context"`	任务 ID
`python_callable`	`print_context`	要执行的 Python 函数
`op_args`	`[1, 2]`	传给函数的位置参数
`op_kwargs`	`{"name": "airflow"}`	传给函数的关键字参数
`templates_dict`	`{"biz_date": "{{ ds }}"}`	模板参数字典
`show_return_value_in_logs`	`True`	是否在日志中显示返回值

6.1 使用 `op_args`

def add(a, b):
    print(a + b)


add_task = PythonOperator(
    task_id="add_task",
    python_callable=add,
    op_args=[1, 2],
)

执行时等价于：

add(1, 2)

6.2 使用 `op_kwargs`

def submit_job(layer, biz_date):
    print(f"submit {layer}, biz_date={biz_date}")


dwd = PythonOperator(
    task_id="submit_dwd",
    python_callable=submit_job,
    op_kwargs={
        "layer": "dwd",
        "biz_date": "{{ ds }}",
    },
)

执行时等价于：

submit_job(layer="dwd", biz_date="2023-06-21")

这里的 {{ ds }} 会在任务运行时渲染。

6.3 PythonOperator 适合做什么

PythonOperator 适合做轻量级逻辑：

生成业务日期。
调用接口。
检查文件是否存在。
拼接下游任务参数。
写少量元数据。
做分支判断。

不建议在 PythonOperator 中直接写大规模计算逻辑。

例如大数据场景中，下面这种方式不太合适：

def calculate_big_data():
    # 直接在 Airflow Worker 中跑大量计算
    ...

更合理的方式是：Airflow 负责任务调度，计算交给 Spark、Hive、Flink 等计算引擎。

例如：

submit_spark = BashOperator(
    task_id="submit_spark",
    bash_command="spark-submit ...",
)

或者使用对应 provider 中的 Operator。

七、常见 Operator 类型

Airflow 中 Operator 表示一个具体任务。

不同 Operator 负责不同类型的操作。

常见类型如下：

Operator	作用	典型场景
`BashOperator`	执行 Bash 命令	Shell、Spark Submit、Hive 命令
`PythonOperator`	执行 Python 函数	参数处理、接口调用、轻量逻辑
`BranchPythonOperator`	根据条件选择下游分支	成功走一条链路，失败走另一条链路
`ShortCircuitOperator`	条件不满足时跳过下游任务	数据不存在时跳过后续处理
`EmptyOperator`	空任务，占位使用	开始节点、结束节点、分组连接
`EmailOperator`	发送邮件	任务完成通知
`TriggerDagRunOperator`	触发另一个 DAG	DAG 之间依赖
`Sensor`	等待某个条件满足	等文件、等时间、等外部任务

7.1 `EmptyOperator`

EmptyOperator 本身不做实际工作，通常用于占位。

例如：

from airflow.operators.empty import EmptyOperator

start = EmptyOperator(task_id="start")
end = EmptyOperator(task_id="end")

start >> dwd >> dws >> ads >> end

这样在 Graph View 中会更清晰：

start -> dwd -> dws -> ads -> end

7.2 `BranchPythonOperator`

BranchPythonOperator 用来做分支。

例如根据业务日期判断走哪条链路：

from airflow.operators.python import BranchPythonOperator


def choose_path(**context):
    ds = context["ds"]
    if ds.endswith("-01"):
        return "monthly_task"
    return "daily_task"


branch = BranchPythonOperator(
    task_id="choose_path",
    python_callable=choose_path,
)

它返回的是下游任务的 task_id。

如果返回：

return "daily_task"

那么 daily_task 会继续执行，其他未选中的分支会被标记为 skipped。

7.3 `ShortCircuitOperator`

ShortCircuitOperator 用来判断后续任务是否继续执行。

例如判断数据文件是否存在：

from airflow.operators.python import ShortCircuitOperator


def check_data_file():
    import os
    return os.path.exists("/tmp/data.txt")


check = ShortCircuitOperator(
    task_id="check_data_file",
    python_callable=check_data_file,
)

如果函数返回 True，下游任务继续执行。

如果返回 False，下游任务会被跳过。

这种写法适合处理“没有数据就不继续跑”的场景。

八、任务依赖怎么写

Airflow 中最常见的依赖写法是：

dwd >> dws >> ads

表示：

dwd -> dws -> ads

也可以反过来写：

ads << dws << dwd

这两种写法表达的依赖关系一样。

8.1 一个任务依赖多个上游

[dwd_user, dwd_order] >> dws_trade

表示：

dwd_user  \
           -> dws_trade
dwd_order /

也就是 dws_trade 要等 dwd_user 和 dwd_order 都完成后才能执行。

8.2 一个任务后接多个下游

dwd >> [dws_user, dws_order]

表示：

        -> dws_user
dwd
        -> dws_order

dwd 成功后，两个下游任务都可以执行。

8.3 使用 `chain`

如果任务比较多，可以使用 chain。

from airflow.models.baseoperator import chain

chain(dwd, dws, ads)

效果等价于：

dwd >> dws >> ads

如果链路很长，chain 会比一长串 >> 更清楚。

8.4 使用 `cross_downstream`

如果有多个上游任务和多个下游任务都要建立依赖，可以使用 cross_downstream。

from airflow.models.baseoperator import cross_downstream

cross_downstream([dwd_user, dwd_order], [dws_user, dws_order])

表示两个上游任务都分别指向两个下游任务。

但是这种写法要谨慎使用。

如果依赖关系太密，Graph View 会变得很乱。

普通 DAG 中优先使用 >> 和 chain。

几种任务依赖写法对比图

这张图中需要重点理解的是：不同依赖写法最后都会变成 DAG 图中的边。

写 DAG 时不要只看 Python 代码是否简短，还要考虑最终在 Graph View 中是否清晰。如果依赖关系过于密集，后续排查任务状态会比较困难。

九、`trigger_rule` 触发规则

默认情况下，一个任务只有在所有上游任务成功后才会执行。

这个默认规则可以理解为：

trigger_rule="all_success"

常见触发规则如下：

触发规则	含义
`all_success`	所有上游任务成功后执行，默认值
`all_failed`	所有上游任务失败后执行
`all_done`	所有上游任务都结束后执行，不管成功失败
`one_success`	至少一个上游任务成功后执行
`one_failed`	至少一个上游任务失败后执行
`none_failed`	没有上游任务失败即可执行
`none_skipped`	没有上游任务被跳过即可执行

例如有一个结束通知任务，希望不管上游成功还是失败都执行：

notify = BashOperator(
    task_id="notify",
    bash_command="echo 'dag finished'",
    trigger_rule="all_done",
)

这种场景下，如果继续使用默认的 all_success，一旦上游任务失败，通知任务也不会执行。

所以写复杂 DAG 时，要特别注意 trigger_rule。

十、模板变量

Airflow 支持 Jinja 模板。

在 BashOperator 中，bash_command 和 env 都是可以模板渲染的字段。

例如：

dwd = BashOperator(
    task_id="dwd",
    bash_command="""
    echo "ds={{ ds }}"
    echo "ts={{ ts }}"
    echo "dag_id={{ dag.dag_id }}"
    echo "task_id={{ task.task_id }}"
    """,
)

常见模板变量如下：

变量	说明
`{{ ds }}`	逻辑日期，格式为 `YYYY-MM-DD`
`{{ ds_nodash }}`	逻辑日期，格式为 `YYYYMMDD`
`{{ ts }}`	逻辑时间戳
`{{ dag.dag_id }}`	当前 DAG ID
`{{ task.task_id }}`	当前任务 ID
`{{ run_id }}`	当前 DAG Run ID
`{{ dag_run }}`	当前 DAG Run 对象
`{{ params }}`	参数对象

大数据任务中最常用的是业务日期。

例如：

dwd = BashOperator(
    task_id="dwd",
    bash_command="""
    spark-submit \
      --class com.example.DwdJob \
      /opt/jobs/dwd.jar \
      --biz-date {{ ds }}
    """,
)

如果 Hive 分区使用 yyyyMMdd，可以使用：

bash_command="echo {{ ds_nodash }}"

输出类似：

20230621

十一、XCom 与返回值

XCom 用来在任务之间传递少量数据。

BashOperator 在开启 do_xcom_push=True 时，会把标准输出的最后一行推送到 XCom。

PythonOperator 的函数返回值，也可以进入 XCom。

例如：

def get_biz_date(**context):
    return context["ds"]


get_date = PythonOperator(
    task_id="get_biz_date",
    python_callable=get_biz_date,
)

但是要注意：

XCom 不适合传递大数据。

不要把大量结果数据、DataFrame、文件内容放到 XCom 中。

它更适合传递：

一个日期。
一个文件路径。
一个状态标记。
一个小的 JSON 参数。

真正的大数据结果应该写到 HDFS、Hive、MySQL、对象存储等外部系统中。

十二、任务级别常用参数

不管是 BashOperator 还是 PythonOperator，它们都继承自 BaseOperator。

所以很多参数是通用的。

常见任务级别参数如下：

参数	示例	说明
`task_id`	`"dwd"`	任务唯一标识
`owner`	`"test_owner"`	负责人
`retries`	`3`	重试次数
`retry_delay`	`timedelta(minutes=5)`	重试间隔
`depends_on_past`	`False`	是否依赖上一个周期的同名任务
`execution_timeout`	`timedelta(hours=1)`	单次执行最大时长
`trigger_rule`	`"all_success"`	上游任务状态满足什么条件才执行
`pool`	`"default_pool"`	使用哪个资源池
`pool_slots`	`1`	占用几个 pool slot
`priority_weight`	`1`	调度优先级权重
`queue`	`"default"`	队列名称，CeleryExecutor 中常用
`do_xcom_push`	`True`	是否将结果推送到 XCom
`on_failure_callback`	`func`	失败回调
`on_success_callback`	`func`	成功回调
`on_retry_callback`	`func`	重试回调

12.1 `execution_timeout`

如果一个任务长时间卡住，可以设置超时时间。

dwd = BashOperator(
    task_id="dwd",
    bash_command="sleep 3600",
    execution_timeout=timedelta(minutes=10),
)

如果任务超过 10 分钟还没有结束，会被 Airflow 标记为失败。

12.2 `pool`

pool 用来控制资源并发。

例如某个外部系统最多只能同时承受 3 个任务，可以创建一个 pool，然后让相关任务都使用这个 pool。

dwd = BashOperator(
    task_id="dwd",
    bash_command="echo dwd",
    pool="spark_pool",
)

这样可以避免大量任务同时提交，把外部系统打满。

12.3 回调函数

任务失败时可以执行回调函数。

例如：

def notify_failure(context):
    dag_id = context["dag"].dag_id
    task_id = context["task"].task_id
    print(f"task failed: {dag_id}.{task_id}")


dwd = BashOperator(
    task_id="dwd",
    bash_command="exit 1",
    on_failure_callback=notify_failure,
)

这里先用 print 演示。

实际生产中可以在回调中发送邮件、企业微信、钉钉或者写入告警系统。

十三、编写 DAG 时的几个注意点

13.1 DAG 文件顶层不要写重逻辑

Airflow Scheduler 会周期性解析 DAG 文件。

所以 DAG 文件的顶层代码应该尽量轻量。

不建议在顶层写：

# 不建议
data = requests.get("http://example.com/api").json()

也不建议在顶层做大量数据库查询。

这些逻辑应该放到具体 Operator 中执行。

否则 Scheduler 每次解析 DAG 文件都会执行这些逻辑，容易导致 DAG 解析变慢，甚至影响调度。

13.2 `start_date` 不要写 `datetime.now()`

不建议这样写：

start_date=datetime.now()

因为 Scheduler 每次解析 DAG 文件时，datetime.now() 都会变化。

这会让调度逻辑变得不可预测。

建议写成固定时间：

start_date=datetime(2023, 6, 21)

13.3 `task_id` 保持稳定

如果任务已经运行过，不建议随便修改 task_id。

因为 task_instance 表中会按照 dag_id、task_id、run_id 记录任务实例。

修改 task_id 后，Airflow 会认为这是一个新的任务。

13.4 不要让 Airflow Worker 做大计算

Airflow 的定位是调度和编排。

它不是 Spark，也不是 Flink。

所以大数据任务中，Airflow 更适合做：

调度 Spark 任务
调度 Hive SQL
调度 Shell 脚本
调度接口调用

而不是把大量计算代码直接写在 Worker 进程中。

13.5 Bash 命令要关注退出码

Airflow 判断 BashOperator 成功或失败，关键看 Bash 进程的退出码。

所以多命令脚本中建议加：

set -e

如果只写：

bash_command="python a.py; echo done"

即使 python a.py 失败，后面的 echo done 成功，也可能让最终退出码变成 0。

这种问题排查起来很麻烦。

十四、一个更完整的 DAG 示例

下面整理一个稍微完整一点的写法：

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator


default_args = {
    "owner": "test_owner",
    "depends_on_past": False,
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}


with DAG(
    dag_id="spark_layer_demo_v2",
    default_args=default_args,
    description="Spark layer dependency demo",
    schedule_interval="@daily",
    start_date=datetime(2023, 6, 21),
    catchup=False,
    tags=["spark", "demo"],
    max_active_runs=1,
) as dag:

    start = EmptyOperator(task_id="start")

    dwd = BashOperator(
        task_id="dwd",
        bash_command="""
        set -e
        echo "submit dwd spark job, biz_date={{ ds }}"
        date
        """,
        retries=3,
        retry_delay=timedelta(minutes=3),
        execution_timeout=timedelta(minutes=30),
    )

    dws = BashOperator(
        task_id="dws",
        bash_command="""
        set -e
        echo "submit dws spark job, biz_date={{ ds }}"
        date
        """,
        retries=3,
        retry_delay=timedelta(minutes=3),
        execution_timeout=timedelta(minutes=30),
    )

    ads = BashOperator(
        task_id="ads",
        bash_command="""
        set -e
        echo "submit ads spark job, biz_date={{ ds }}"
        date
        """,
        retries=3,
        retry_delay=timedelta(minutes=3),
        execution_timeout=timedelta(minutes=30),
    )

    end = EmptyOperator(
        task_id="end",
        trigger_rule="all_done",
    )

    start >> dwd >> dws >> ads >> end

相比最开始的版本，这个 DAG 多了几个点：

增加了 start 和 end 占位任务。
schedule_interval 使用了 @daily。
设置了 max_active_runs=1，避免同一个 DAG 同时跑多个周期。
Bash 命令中增加了 set -e。
使用 {{ ds }} 作为业务日期。
单个任务增加了 execution_timeout。
结束任务使用 trigger_rule="all_done"。

这种写法更接近实际使用。

十五、验证 DAG 是否正常

写完 DAG 文件后，可以先在命令行执行 Python 文件，检查语法错误：

$ python /home/parallels/airflow234/dags/spark_layer_demo_v2.py

如果没有报错，再查看 Airflow 是否识别：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow dags list

查看任务：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow tasks list spark_layer_demo_v2

查看依赖树：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow tasks list spark_layer_demo_v2 --tree

测试单个任务：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow tasks test spark_layer_demo_v2 dwd 2023-07-05

如果 DAG 没有出现，可以查看导入错误：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow dags list-import-errors

或者查询元数据库：

SELECT timestamp, filename, stacktrace
FROM import_error
ORDER BY timestamp DESC;

十六、总结

这一篇主要围绕 DAG 代码本身做了整理。

需要重点记住几点：

DAG 文件本质是 Python 文件，但 Airflow 只会加载顶层的 DAG 对象。
DAG(...) 定义工作流，Operator(...) 定义具体任务。
default_args 用来减少重复配置，但任务自己的参数优先级更高。
BashOperator 适合执行 Shell、Spark Submit、Hive 命令等。
PythonOperator 适合执行轻量级 Python 逻辑，不适合承载大规模计算。
任务依赖推荐使用 >>、<<、chain。
复杂场景中要注意 trigger_rule，否则上游失败或跳过后，下游任务可能不会执行。
大数据任务中经常使用 {{ ds }}、{{ ds_nodash }} 作为业务日期。
多命令 Bash 任务建议加 set -e，避免中间命令失败但任务仍然成功。
写完 DAG 后先用命令行验证，再去 Web UI 中观察状态。

到这里，Airflow 的 DAG 编写方式就比较完整了。

后续再写复杂调度任务时，可以继续围绕分支任务、Sensor、XCom、跨 DAG 依赖、失败告警等内容展开。

参考资料

hnbian

https://www.hnbian.cn/posts/f7a33e63.html

本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !

airflow dag operator

Airflow 系列7：运维与问题排查

2023-07-12 airflow

airflow ops troubleshooting

对 kafka 中的消息进行 Avro 序列化

2023-07-02 kafka

kafka avro

Airflow 系列6：DAG 代码编写与 Operator 参数详解

一、先看完整 DAG 示例

二、导包方式

三、default_args 的作用

四、DAG(...) 常用参数

4.1 dag_id

4.2 schedule_interval

4.3 start_date

4.4 catchup

五、BashOperator 参数详解

5.1 task_id

5.2 bash_command

5.3 建议加 set -e

5.4 env 和模板变量

5.5 skip_exit_code

六、PythonOperator 参数详解

6.1 使用 op_args

6.2 使用 op_kwargs

6.3 PythonOperator 适合做什么

七、常见 Operator 类型

7.1 EmptyOperator

7.2 BranchPythonOperator

7.3 ShortCircuitOperator

八、任务依赖怎么写

8.1 一个任务依赖多个上游

8.2 一个任务后接多个下游

8.3 使用 chain

8.4 使用 cross_downstream

九、trigger_rule 触发规则

十、模板变量

十一、XCom 与返回值

十二、任务级别常用参数

12.1 execution_timeout

12.2 pool

12.3 回调函数

十三、编写 DAG 时的几个注意点

13.1 DAG 文件顶层不要写重逻辑

13.2 start_date 不要写 datetime.now()

13.3 task_id 保持稳定

13.4 不要让 Airflow Worker 做大计算

13.5 Bash 命令要关注退出码

十四、一个更完整的 DAG 示例

十五、验证 DAG 是否正常

十六、总结

参考资料

三、`default_args` 的作用

四、`DAG(...)` 常用参数

4.1 `dag_id`

4.2 `schedule_interval`

4.3 `start_date`

4.4 `catchup`

五、`BashOperator` 参数详解

5.1 `task_id`

5.2 `bash_command`

5.3 建议加 `set -e`

5.4 `env` 和模板变量

5.5 `skip_exit_code`

六、`PythonOperator` 参数详解

6.1 使用 `op_args`

6.2 使用 `op_kwargs`

7.1 `EmptyOperator`

7.2 `BranchPythonOperator`

7.3 `ShortCircuitOperator`

8.3 使用 `chain`

8.4 使用 `cross_downstream`

九、`trigger_rule` 触发规则

12.1 `execution_timeout`

12.2 `pool`

13.2 `start_date` 不要写 `datetime.now()`

13.3 `task_id` 保持稳定