Airflow 系列5：DAG 编写与任务调度使用介绍

airflow

发布日期: 2023-06-21

文章字数: 3.7k

阅读时长: 15 分

阅读次数:

前面几篇文章已经完成了 Airflow 的基本概念介绍、单机部署、元数据库配置以及元数据库表结构介绍。

到这里，Airflow 已经具备了基本使用条件：

Webserver 可以访问。
Scheduler 可以正常运行。
元数据库已经切换到 MySQL。
Executor 已经修改为 LocalExecutor。

接下来就可以开始真正编写 DAG 任务。

本文基于 Apache Airflow 2.3.4，主要介绍 DAG 文件如何编写、任务依赖如何设置、如何在命令行和页面上查看 DAG、如何触发任务以及如何查看任务日志。

一、DAG 文件放在哪里

Airflow 会从配置文件中的 dags_folder 目录扫描 DAG 文件。

可以通过下面的命令查看当前配置：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow config get-value core dags_folder

如果没有单独修改，通常会在 $AIRFLOW_HOME/dags 目录下。

在当前环境中，可以创建 DAG 目录：

$ mkdir -p /home/parallels/airflow234/dags
$ cd /home/parallels/airflow234/dags

Word 文档中的操作也是先创建 dags 目录，然后将 Python 脚本放到这个目录：

(airflow) [atguigu@hadoop102 airflow]$ mkdir ~/airflow/dags
(airflow) [atguigu@hadoop102 airflow]$ cd dags/
(airflow) [atguigu@hadoop102 dags]$ vim test.py

这里需要注意：

Airflow 的 DAG 文件本质上就是 Python 文件，但是不是所有 Python 文件都会被识别成 DAG。

只有文件中定义了 DAG 对象，并且 Airflow 能够正常解析，页面上才会出现对应的 DAG。

二、一个最简单的 DAG 结构

先看一个简单示例。

from datetime import datetime, timedelta

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


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


with DAG(
    dag_id="demo_bash_dag",
    default_args=default_args,
    description="A simple bash dag demo",
    schedule_interval=timedelta(days=1),
    start_date=datetime(2023, 6, 21),
    catchup=False,
    tags=["demo"],
) as dag:

    t1 = BashOperator(
        task_id="print_date",
        bash_command="date",
    )

    t2 = BashOperator(
        task_id="sleep",
        bash_command="sleep 5",
    )

    t3 = BashOperator(
        task_id="echo",
        bash_command="echo 'hello airflow'",
    )

    t1 >> t2 >> t3

这个 DAG 中有三个任务：

print_date -> sleep -> echo

每个任务都是一个 BashOperator，表示执行一段 Shell 命令。

最后一行：

t1 >> t2 >> t3

表示任务依赖关系。

也就是说，只有 t1 成功后，t2 才会执行；只有 t2 成功后，t3 才会执行。

三、Airflow 2.3.4 中的导包方式

Word 文档中的示例使用了旧版本导包方式：

from airflow.operators.bash_operator import BashOperator

在 Airflow 2.3.4 中，更推荐使用下面这种方式：

from airflow.operators.bash import BashOperator

如果要使用 Python 函数作为任务，可以这样导入：

from airflow.operators.python import PythonOperator

旧的导包方式在某些版本中仍然可能兼容，但是新文章中建议按 Airflow 2.x 的写法来写。

四、DAG 参数说明

在 DAG 中，几个参数比较重要。

with DAG(
    dag_id="demo_bash_dag",
    default_args=default_args,
    schedule_interval=timedelta(days=1),
    start_date=datetime(2023, 6, 21),
    catchup=False,
    tags=["demo"],
) as dag:
    ...

4.1 `dag_id`

dag_id 是 DAG 的唯一标识。

页面展示、命令行操作、元数据库查询，基本都会用到这个 ID。

例如：

$ airflow dags list
$ airflow tasks list demo_bash_dag

都需要通过 dag_id 找到对应 DAG。

4.2 `start_date`

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

它不是“DAG 文件创建时间”，也不是“Airflow 第一次看到这个 DAG 的时间”。

例如：

start_date=datetime(2023, 6, 21)

表示从这个逻辑时间之后，Airflow 才会根据调度周期创建 DAG Run。

start_date 设置不合理时，经常会导致 DAG 看起来没有按预期调度。

4.3 `schedule_interval`

schedule_interval 表示调度周期。

常见写法如下：

schedule_interval=timedelta(days=1)

也可以使用 Cron 表达式：

schedule_interval="0 2 * * *"

表示每天凌晨 2 点调度。

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

schedule_interval=None

4.4 `catchup`

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

如果写：

catchup=True

当 start_date 很早，而当前时间已经过去很久，Airflow 可能会自动创建很多历史 DAG Run。

学习和测试阶段，通常建议设置：

catchup=False

这样 Airflow 不会自动补跑大量历史任务。

4.5 `default_args`

default_args 用来统一设置任务默认参数。

例如：

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "email": ["example@example.com"],
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

常见参数说明如下：

参数	说明
`owner`	任务所属用户
`depends_on_past`	是否依赖上一次任务实例成功
`email`	邮件接收人
`email_on_failure`	失败是否发送邮件
`email_on_retry`	重试是否发送邮件
`retries`	失败后重试次数
`retry_delay`	重试间隔

其中 depends_on_past 容易误解。

它不是设置当前 DAG 中任务之间的依赖关系，而是表示当前 Task Instance 是否依赖上一个调度周期的同一个 Task Instance 成功。

普通场景下，一般先设置为：

"depends_on_past": False

五、使用 BashOperator 编排 Spark 示例

Word 文档中给了一个比较贴近大数据场景的示例：

dwd -> dws -> ads

每个任务通过 spark-submit 提交一个 Spark 官方示例。

这里需要先说明一个前提：

Airflow 只是负责调度任务，不负责替我们启动 Hadoop、YARN 或 Spark 集群。

如果 DAG 里要通过 spark-submit --master yarn 提交任务，那么测试前需要先保证 Hadoop、YARN 和 Spark 相关服务已经启动。

Word 文档中的测试环境是先启动 Hadoop 和 Spark History Server：

$ myhadoop.sh start
$ /opt/module/spark-yarn/sbin/start-history-server.sh

如果这些外部服务没有启动，Airflow 任务本身可以被调度起来，但是 spark-submit 命令会失败。这个失败属于外部计算环境问题，不是 Airflow 调度器本身的问题。

下面按照 Airflow 2.3.4 的写法整理一下：

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": ["example@example.com"],
    "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,
    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=(
            '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"'
        ),
        retries=3,
    )

    dws = BashOperator(
        task_id="dws",
        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"'
        ),
        retries=3,
    )

    ads = BashOperator(
        task_id="ads",
        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"'
        ),
        retries=3,
    )

    dwd >> dws >> ads

这个示例表达的是数仓分层任务依赖：

dwd -> dws -> ads

也就是：

先执行 dwd。
dwd 成功后执行 dws。
dws 成功后执行 ads。

如果 dwd 失败，后面的 dws 和 ads 默认不会继续执行。

六、使用 PythonOperator 编写 Python 任务

除了执行 Shell 命令，也可以执行 Python 函数。

例如：

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.python import PythonOperator


def extract_data():
    print("extract data")


def transform_data():
    print("transform data")


def load_data():
    print("load data")


with DAG(
    dag_id="python_operator_demo",
    start_date=datetime(2023, 6, 21),
    schedule_interval=None,
    catchup=False,
    default_args={
        "owner": "airflow",
        "retries": 1,
        "retry_delay": timedelta(minutes=5),
    },
) as dag:

    extract = PythonOperator(
        task_id="extract",
        python_callable=extract_data,
    )

    transform = PythonOperator(
        task_id="transform",
        python_callable=transform_data,
    )

    load = PythonOperator(
        task_id="load",
        python_callable=load_data,
    )

    extract >> transform >> load

这个示例中，python_callable 指定要执行的 Python 函数。

适合放一些轻量级的调度逻辑、接口调用、参数处理等。

如果任务本身是大规模计算，不建议直接在 PythonOperator 中执行大量计算逻辑，更合理的方式是提交到 Spark、Hive、Flink 或其他计算引擎中执行。

七、任务依赖写法

Airflow 中常用的依赖写法有几种。

7.1 使用 `>>`

t1 >> t2 >> t3

表示：

t1 -> t2 -> t3

7.2 使用 `<<`

t3 << t2 << t1

含义和下面一样：

t1 >> t2 >> t3

7.3 一个任务依赖多个上游

[t1, t2] >> t3

表示 t3 依赖 t1 和 t2。

只有 t1、t2 都满足条件后，t3 才会执行。

7.4 一个任务后面接多个下游

t1 >> [t2, t3]

表示 t1 成功后，t2 和 t3 都可以执行。

7.5 使用 `set_upstream`

Word 文档中使用的是：

t2.set_upstream(t1)
t3.set_upstream(t2)

含义是：

t1 -> t2 -> t3

这种写法没有问题，只是现在更常见的是使用 >>，可读性更强。

八、让 Airflow 识别 DAG

把 DAG 文件放到 dags 目录后，Scheduler 会周期性扫描。

可以先检查 DAG 文件是否有语法错误：

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

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

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

Word 文档中的命令是：

(airflow) [atguigu@hadoop102 airflow]$ airflow dags list

如果 DAG 已经被识别，会在列表中看到对应的 dag_id。

例如：

spark_layer_demo

在 Web 页面中也可以看到这个 DAG：

spark_layer_demo DAG 页面

从这个页面中可以看到几个信息：

DAG 名称是 spark_layer_demo。
DAG 当前处于开启状态。
调度周期是每天执行一次。
DAG 中包含 3 个任务，分别是 dwd、dws、ads。
当前页面主要用来确认 DAG 已经被 Airflow 识别，还没有重点查看任务执行结果。

如果没有看到，常见原因包括：

DAG 文件没有放到 dags_folder。
Python 文件有语法错误。
导包失败，例如某个第三方包没有安装。
文件中没有定义 DAG 对象。
Scheduler 还没有完成下一轮解析。

可以查看导入错误：

$ 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 下有哪些任务：

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

如果想按依赖树展示：

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

Word 文档中也提到了类似命令：

# 查看单个任务
(airflow) [atguigu@hadoop102 airflow]$ airflow tasks list test --tree

输出类似：

<Task(BashOperator): dwd>
    <Task(BashOperator): dws>
        <Task(BashOperator): ads>

通过这个命令，可以快速确认代码里的依赖关系是否和预期一致。

十、手动触发 DAG

在页面上可以点击 DAG 右侧的运行按钮手动触发。

DAG 页面右侧的手动触发按钮

命令行也可以触发：

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

触发之后，可以查看 DAG Run：

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

也可以直接查元数据库：

SELECT dag_id,
       run_id,
       run_type,
       state,
       execution_date,
       start_date,
       end_date
FROM dag_run
WHERE dag_id = 'spark_layer_demo'
ORDER BY execution_date DESC;

十一、查看任务日志

Airflow 页面上可以点击具体 Task Instance，然后进入日志页面查看任务日志。

Word 文档中也提到：

点击成功任务，查看日志

命令行也可以查看任务日志：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow tasks logs spark_layer_demo dwd <run_id>

其中：

spark_layer_demo  是 dag_id
dwd               是 task_id
<run_id>          是本次 DAG Run 的 run_id

如果不确定 run_id，可以先查：

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

或者查 MySQL：

SELECT run_id, state, execution_date
FROM dag_run
WHERE dag_id = 'spark_layer_demo'
ORDER BY execution_date DESC;

十二、Web UI 中常用视图

Airflow 页面上有几个视图非常常用。

12.1 Grid View

Grid View 用来查看 DAG Run 和 Task Instance 的状态矩阵。

它适合快速判断：

哪一次 DAG Run 失败了。
哪个 Task 失败了。
某个任务是否一直处于 queued 或 running。

Grid View 中的任务状态

12.2 Graph View

Graph View 用来查看任务依赖关系。

它适合确认 DAG 编写出来的依赖是否符合预期。

例如：

dwd -> dws -> ads

在 Graph View 中应该表现为一条从左到右或从上到下的任务链路。

spark_layer_demo 的 Graph View 页面

12.3 Gantt View

Gantt View 用来查看任务执行耗时。

它适合分析：

哪个任务耗时最长。
哪些任务是串行执行的。
哪些任务可以并行优化。

12.4 Code View

Code View 用来查看 DAG 文件代码。

如果页面上看到的代码不是自己预期的版本，就要检查：

DAG 文件是否保存到了正确目录。
Scheduler 是否已经重新解析。
是否存在多个同名 dag_id 的文件。

Code View 中的部分 DAG 源码

十三、失败重试配置

Airflow 中任务失败后是否重试，主要由 retries 和 retry_delay 控制。

例如：

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

表示任务失败后最多重试 3 次，每次间隔 5 分钟。

也可以在单个 Operator 上覆盖：

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

需要注意：

重试不是无限重跑。

如果任务失败后进入 up_for_retry 状态，说明它正在等待下一次重试时间。

可以在元数据库中查看：

SELECT dag_id,
       task_id,
       run_id,
       state,
       try_number,
       start_date,
       end_date
FROM task_instance
WHERE state IN ('failed', 'up_for_retry')
ORDER BY start_date DESC;

十四、一个 DAG 没有出现时怎么排查

如果写完 DAG 文件后，页面中没有看到这个 DAG，可以按下面顺序排查。

14.1 确认文件目录

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow config get-value core dags_folder

确认文件确实在这个目录下。

14.2 确认 Python 文件可以执行

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

如果这里就报错，Airflow 解析时也会失败。

14.3 查看导入错误

$ 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;

14.4 确认 Scheduler 是否运行

$ ps -ef | grep airflow | grep scheduler

也可以查看健康接口：

$ curl -fsS http://127.0.0.1:8080/health

如果 Scheduler 没有运行，DAG 文件不会正常被调度解析。

十五、常用命令汇总

查看所有 DAG：

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

查看导入错误：

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

查看某个 DAG 的任务：

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

按树形结构查看任务：

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

手动触发 DAG：

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

查看 DAG Run：

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

测试某个任务：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow tasks test spark_layer_demo dwd 2023-06-21

查看 Airflow 版本：

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

十六、总结

本文主要介绍了 Airflow DAG 的基本编写和使用方式。

需要重点记住几点：

DAG 文件要放在 dags_folder 目录下。
一个 DAG 文件本质是 Python 文件，但必须能被 Airflow 正常解析。
dag_id 是 DAG 的唯一标识，后续命令行和元数据库查询都会用到。
start_date、schedule_interval、catchup 会影响 DAG 是否自动调度。
BashOperator 适合执行 Shell 命令，例如提交 Spark 任务。
PythonOperator 适合执行 Python 函数。
任务依赖推荐使用 >> 这种写法，简单直观。
DAG 是否识别可以看 airflow dags list，任务依赖可以看 airflow tasks list --tree。
DAG 运行状态最终会进入 dag_run，任务运行状态最终会进入 task_instance。

到这里，Airflow 已经从部署环境进入了实际 DAG 编写阶段。

下一篇文章可以继续整理 Airflow 的运维和问题排查，例如 DAG 不显示、任务一直 queued、Scheduler 没有心跳、任务失败重试、日志查看和元数据库排查 SQL。

hnbian

https://www.hnbian.cn/posts/4805e5ed.html

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

airflow dag

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

2023-07-02 kafka

kafka avro

Airflow 系列4：Airflow 元数据库表结构介绍

2023-06-15 airflow

airflow mysql

Airflow 系列5：DAG 编写与任务调度使用介绍

一、DAG 文件放在哪里

二、一个最简单的 DAG 结构

三、Airflow 2.3.4 中的导包方式

四、DAG 参数说明

4.1 dag_id

4.2 start_date

4.3 schedule_interval

4.4 catchup

4.5 default_args

五、使用 BashOperator 编排 Spark 示例

六、使用 PythonOperator 编写 Python 任务

七、任务依赖写法

7.1 使用 >>

7.2 使用 <<

7.3 一个任务依赖多个上游

7.4 一个任务后面接多个下游

7.5 使用 set_upstream

八、让 Airflow 识别 DAG

九、查看 DAG 中的任务

十、手动触发 DAG

十一、查看任务日志

十二、Web UI 中常用视图

12.1 Grid View

12.2 Graph View

12.3 Gantt View

12.4 Code View

十三、失败重试配置

十四、一个 DAG 没有出现时怎么排查

14.1 确认文件目录

14.2 确认 Python 文件可以执行

14.3 查看导入错误

14.4 确认 Scheduler 是否运行

十五、常用命令汇总

十六、总结

4.1 `dag_id`

4.2 `start_date`

4.3 `schedule_interval`

4.4 `catchup`

4.5 `default_args`

7.1 使用 `>>`

7.2 使用 `<<`

7.5 使用 `set_upstream`