第19章 Airflow基础

19、Airflow基础：Airflow架构、DAG定义、Operator与Task、调度配置

说实话，Airflow 是我在数据工程生涯中，用得最顺手的调度工具之一。它不像某些老牌调度器那样死板，也不像某些新框架那样花里胡哨。它就是一个很实在的「工作流编排引擎」。今天咱们就来拆解一下它的核心。

19.1 Airflow 整体架构：谁在干活？

Airflow 不是单打独斗的。它内部有好几个组件，各司其职。我刚开始接触时，总觉得它很重，后来才明白，正是这种分工，才保证了生产环境的稳定。

核心组件就这几个：

Scheduler（调度器）：它是大脑。负责解析 DAG，判断哪些任务该触发，然后丢给 Worker 去执行。
Worker（执行器）：它是手脚。真正跑 Python 代码、执行 SQL 的地方。
Web Server（Web 服务器）：它是脸面。提供 UI 界面，让你能看到 DAG 跑得怎么样，有没有报错。
Database（元数据库）：它是记忆。记录 DAG 的状态、任务实例、变量、连接信息等。默认是 SQLite，生产环境我建议换成 PostgreSQL 或 MySQL。

它们之间的关系，我画了一张图，你看一眼就明白了：

我的小建议： 刚开始学习时，可以在本地用 airflow standalone 命令一键启动所有组件。它会自动初始化数据库、创建用户，省去很多配置烦恼。

19.2 DAG 定义：你的工作流蓝图

DAG（有向无环图）是 Airflow 的核心概念。说白了，它就是一张任务执行顺序的蓝图。每个 DAG 就是一个 Python 文件，放在 dags/ 目录下。

我见过不少新手把 DAG 写得又臭又长，其实核心就三要素：

DAG 对象：定义名称、调度间隔、开始时间等。
Task 对象：定义具体要做什么事。
依赖关系：用 >> 或 set_downstream() 指定谁先谁后。

来看一个最简例子：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator

def print_hello():
    print("Hello, Airflow!")

default_args = {
    'owner': 'data_team',
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id='my_first_dag',
    default_args=default_args,
    description='一个简单的示例 DAG',
    schedule_interval='0 8 * * *',  # 每天早上8点跑
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:

    task_hello = PythonOperator(
        task_id='say_hello',
        python_callable=print_hello,
    )

    task_hello

  注意： catchup=False 这个参数很重要。如果设为 True，当你部署一个过去开始时间的 DAG 时，它会自动把从 start_date 到现在的所有调度周期都补跑一遍。我曾在生产环境吃过这个亏，一下子补跑了三个月的数据，把数据库都压垮了。

19.3 Operator 与 Task：干活的人与活本身

这两个概念容易混淆。我简单解释一下：

Operator 是「模板」，定义了做什么类型的操作。比如 PythonOperator 就是执行 Python 函数，BashOperator 就是执行 Shell 命令。
Task 是「实例」，是 Operator 在 DAG 中被实例化后的具体任务。

打个比方：PythonOperator 就像一张菜谱，而 task_hello 就是按照这张菜谱做出来的一道菜。

常用的 Operator 有这些：

Operator 名称	用途	我常用的场景
`PythonOperator`	执行 Python 函数	数据清洗、调用 API
`BashOperator`	执行 Shell 命令	启动 Spark 任务、文件操作
`PostgresOperator`	执行 SQL 语句	数据加载、表创建
`DummyOperator`	空操作，用于分组	作为分支的汇聚点

依赖关系的写法也很灵活：

# 方式一：用位移运算符
task1 >> task2 >> task3

# 方式二：用链式调用
task1.set_downstream(task2)
task2.set_downstream(task3)

# 方式三：混合使用（我比较喜欢这种）
task1 >> [task2, task3]  # task1 完成后，同时执行 task2 和 task3
task2 >> task4
task3 >> task4

避坑指南： 我曾经在写依赖时，不小心写成了 task1 >> task2 >> task1，形成了一个环。Airflow 在解析 DAG 时会直接报错，说检测到了循环依赖。记住，DAG 里的「A」就是 Acyclic（无环），绝对不能有环！

19.4 调度配置：让 DAG 按时跑起来

调度配置是 Airflow 的灵魂。你想想看，如果任务不能按时自动跑，那还要调度器干嘛？

核心参数就两个：

start_date：DAG 开始生效的时间。注意，它不是第一次执行的时间，而是调度周期的起点。
schedule_interval：调度间隔。可以用 cron 表达式，也可以用 Airflow 预设的常量。

常见的调度写法：

# 每天凌晨2点
schedule_interval='0 2 * * *'

# 每小时执行一次
schedule_interval='0 * * * *'

# 每周一早上8点
schedule_interval='0 8 * * 1'

# 使用预设常量
from airflow.utils.dates import days_ago
schedule_interval='@daily'   # 等同于 0 0 * * *
schedule_interval='@weekly'  # 等同于 0 0 * * 0
schedule_interval='@hourly'  # 等同于 0 * * * *

这里有个容易踩的坑：调度延迟。Airflow 的调度周期是左闭右开的。比如你设置 schedule_interval='@daily'，start_date=datetime(2024,1,1)，那么第一个调度周期是 2024-01-01 00:00:00 到 2024-01-02 00:00:00。这个周期的任务会在 2024-01-02 00:00:00 之后触发。也就是说，任务实际执行时间比周期结束时间晚一点点。

我的习惯： 我会把 start_date 设为一个过去的固定日期，比如 datetime(2024,1,1)，然后配合 catchup=False。这样既保证了调度周期的正确性，又不会因为补跑历史数据而搞崩系统。

最后，别忘了在 Airflow UI 中把 DAG 的开关打开（Toggle 按钮）。我见过有人配置了半天，发现任务没跑，结果一看，DAG 是 Paused 状态。嗯，这种低级错误我也犯过。

好了，Airflow 的基础就这些。架构、DAG、Operator、调度，这四个点吃透了，你就能写出稳定可靠的工作流了。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

19、Airflow基础：Airflow架构、DAG定义、Operator与Task、调度配置

19.1 Airflow 整体架构：谁在干活？

19.2 DAG 定义：你的工作流蓝图

19.3 Operator 与 Task：干活的人与活本身

19.4 调度配置：让 DAG 按时跑起来