Apache Airflow 入门：架构、核心组件与 DAG 部署

说到调度系统，Airflow 几乎是绕不开的选项。我最早接触它是在 2018 年，那时候公司还在用 Cron 加 Shell 脚本硬扛。后来数据量上来，依赖关系一复杂，Cron 根本没法维护。嗯，Airflow 的出现，说白了就是帮你把「什么时候跑、跑什么、跑完怎么办」这三件事理清楚。

这一章，我们聚焦 Airflow 的核心。不扯太深，但该讲的架构、组件、DAG 定义和部署，一个都不能少。

Airflow 的整体架构

Airflow 不是单兵作战。它由几个角色协同工作。我习惯把它理解成一个「编辑部」：

• Scheduler 是主编，负责看稿（检查 DAG）、排期（决定什么时候触发任务）。
• Executor 是发行渠道，决定任务到底在哪台机器上跑。
• Worker 是印刷工，真正干活的人。
• Web Server 是前台，给你看进度、点重跑。
• Metadata Database 是档案室，所有状态都记在里面。

你想想看，如果没有这个档案室，Scheduler 重启后根本不知道任务跑到哪一步了。所以 Metadata DB 是 Airflow 的命根子。

核心组件详解

Scheduler：调度的大脑

Scheduler 是 Airflow 里最忙的组件。它每隔几秒就会扫描 DAG 文件夹，检查哪些 DAG 需要被触发。我个人习惯把 DAG 的 schedule_interval 设成 Cron 表达式，这样更灵活。

Executor：任务的派发策略

Executor 决定了任务怎么跑。Airflow 内置了几种：

Executor 类型	适用场景	特点
SequentialExecutor	本地测试、单机调试	串行执行，一次只跑一个任务。默认使用 SQLite，不能并发。
LocalExecutor	单机多任务并行	利用多进程并行，适合小规模生产。
CeleryExecutor	分布式生产环境	通过 Celery + Redis/RabbitMQ 分发任务，Worker 可以水平扩展。
KubernetesExecutor	云原生、动态扩缩容	每个 Task 启动一个 Pod，用完即销毁。资源隔离性好。

我最早用 LocalExecutor 扛了半年。后来业务量上来，Worker 节点从 1 台变成 5 台，果断切到了 CeleryExecutor。切换过程其实不复杂，主要是把 airflow.cfg 里的 executor 改一下，再配好消息队列就行。

Worker：真正的执行者

Worker 就是干活的。它从消息队列里拉取任务，执行具体的 Operator。Worker 本身是无状态的，你随时可以加机器。嗯，这里要注意：Worker 的 Python 环境和 DAG 依赖必须保持一致，否则任务会报错。

DAG 定义与部署

DAG 是 Airflow 的核心抽象。说白了，它就是一张有向无环图，定义了任务的执行顺序和依赖关系。

一个简单的 DAG 示例

我们来看一个实际例子。假设你要每天凌晨跑一个数据管道：先抽取数据，再清洗，最后加载到数仓。

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

def extract():
    print("抽取数据...")

def transform():
    print("清洗数据...")

def load():
    print("加载数据...")

default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['alerts@company.com'],
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id='etl_pipeline',
    default_args=default_args,
    description='每日 ETL 管道',
    schedule_interval='0 2 * * *',  # 每天凌晨 2 点
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['etl', 'production'],
) as dag:

    t1 = PythonOperator(task_id='extract', python_callable=extract)
    t2 = PythonOperator(task_id='transform', python_callable=transform)
    t3 = PythonOperator(task_id='load', python_callable=load)

    t1 >> t2 >> t3

这段代码定义了一个三步骤的 ETL 流程。t1 >> t2 >> t3 表示 t1 执行完才执行 t2，t2 执行完才执行 t3。这个语法是 Airflow 特有的，用位运算符表示依赖关系。

DAG 部署方式

DAG 写好了，怎么部署到 Airflow 里？有三种常见方式：

1. 直接拷贝到 DAG 文件夹：把 .py 文件放到 $AIRFLOW_HOME/dags/ 目录下。Scheduler 会自动扫描加载。适合小团队。
2. Git 同步：用 Git 管理 DAG 代码，通过 CI/CD 流水线自动同步到服务器。我推荐这种方式，有版本控制，回滚方便。
3. 分布式存储挂载：把 DAG 文件放在 S3、GCS 或 NFS 上，挂载到所有 Scheduler 和 Worker 节点。适合多机房部署。

我个人习惯用 Git 同步。每次提交代码后，CI 自动执行 git pull 到目标服务器，然后触发 Scheduler 重新加载。这样既保证了代码一致性，又方便回滚。

DAG 的调度与执行流程

一个 DAG 从定义到执行，大致经历这几个步骤：

1. 解析：Scheduler 读取 DAG 文件，构建 DAG 对象。
2. 调度：根据 schedule_interval 和 start_date，计算下一次执行时间。
3. 创建 DAG Run：到达调度时间后，Scheduler 在 Metadata DB 中创建一个 DAG Run 实例。
4. 生成 Task Instance：DAG Run 解析内部的任务依赖，生成 Task Instance。
5. 分发执行：Scheduler 把 Task Instance 交给 Executor，Executor 再派发给 Worker。
6. 状态回写：Worker 执行完任务后，把状态（success/failed）写回 Metadata DB。

嗯，这里有个细节：Scheduler 不会直接调用 Worker。它只负责「安排」，真正的执行由 Executor 和 Worker 完成。这种解耦设计让 Airflow 具备了很好的扩展性。

小结

Airflow 的架构其实不复杂。Scheduler 负责调度，Executor 负责分发，Worker 负责执行，Metadata DB 负责记录。DAG 就是你的作业流定义，用 Python 写，部署方式灵活。