Airflow：Apache 数据管道编排实战指南

为什么需要 Airflow：把零散定时任务变成可依赖、可观测的管道

传统的定时任务通常用 cron 写在多台服务器上，每个脚本各跑各的。一旦上游数据没准备好，下游任务照常执行，产生脏数据；任务失败后没有自动重试，只能等人工发现；任务间的依赖关系散落在文档和人的脑子里，换个人就说不清楚。规模稍大（几十个任务、跨团队协作）就会频繁出问题。

Airflow 用“代码即管道”的方式解决：你在 Python 文件里声明 DAG，节点是任务、边是依赖，调度器负责判断“依赖是否就绪、是否到时间”，执行器决定任务在哪里运行，失败自动重试并告警，Web UI 展示每次运行的依赖图、耗时与日志。任务间不再靠“谁先启动”来隐式协调，而是显式声明依赖。

与 Dagster、Prefect 等新一代编排工具相比，Airflow 的气质更偏向“成熟稳定的大生态”：操作符（Operator，即执行特定动作的任务模板）覆盖几乎所有数据系统，社区资料与生产案例最多，是大多数企业数据平台的默认选择；代价是概念多、上手曲线比 Prefect 陡。

安装与环境配置：4 步就绪

步骤一：准备 Python 环境。 Airflow 2.10 支持 Python 3.8–3.12，建议使用独立虚拟环境，避免与系统 Python 或项目依赖互相污染。

步骤二：安装。 官方包名是 apache-airflow，务必携带官方 constraints 文件一起安装。constraints 是官方为每个版本生成的依赖锁定清单，能避免 pandas、SQLAlchemy 等底层依赖版本冲突——这是 Airflow 安装最常踩的坑。

步骤三：初始化。 设置环境变量 AIRFLOW_HOME（默认即 ~/airflow），执行 airflow db migrate 初始化元数据库（2.7 起取代旧的 airflow db init，元数据库用来存放任务实例状态、日志索引与用户账号），然后创建管理员账号。

步骤四：启动与验证。 分别启动 airflow webserver 和 airflow scheduler 两个进程，浏览器访问 http://localhost:8080，用刚创建的账号登录，看到 DAGs 列表即成功。

# 1) 创建并进入 Python 虚拟环境（Python 3.8–3.12）python -m venv airflow-venvsource airflow-venv/bin/activate # 2) 安装（必须携带 constraints，版本号需与本地 Python 版本对应）pip install "apache-airflow==2.10.5" \  --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.10.5/constraints-3.11.txt" # 3) 初始化元数据库并创建管理员账号export AIRFLOW_HOME=~/airflowairflow db migrateairflow users create --username admin --firstname admin --lastname admin \  --role Admin --email admin@example.com --password admin # 4) 启动 Web Server 与 Scheduler（另开终端分别执行）airflow webserver --port 8080airflow scheduler

快速上手：15 分钟跑通第一个 ETL DAG

最小可用闭环的思路是：在 dags 目录放一个 Python 文件 → 调度器周期性扫描该目录并解析出 DAG → 手动触发一次运行 → Web UI 中观察任务是否按依赖顺序执行。整个过程中你不需要写任何服务端代码，只写一个声明依赖的 Python 脚本。

分步操作：

1. 创建 DAG 目录：mkdir -p ~/airflow/dags（若 AIRFLOW_HOME 是自定义路径，则创建 $AIRFLOW_HOME/dags）。

2. 在该目录新建 etl_pipeline.py，写入下方代码。DAG 定义了一个三条任务的链式依赖：抽取 → 转换 → 加载。

3. 验证解析：执行 airflow dags list，看到 etl_pipeline 即表示文件被正确解析；也可在 Web UI 的 DAGs 页面刷新确认。

4. 触发运行：在 DAGs 页面打开该 DAG，把开关置为开启（Pause 状态解除），点击“播放”按钮手动触发一次；或命令行执行 airflow dags trigger etl_pipeline。

5. 查看结果：进入该 DAG 的 Grid/Graph 视图，点击任意任务节点可查看运行日志、耗时与重试记录。

进阶技巧：单个任务调试用 airflow tasks test etl_pipeline transform 2025-01-01，它只运行指定任务不触发上下游；批量生成相似任务用 TaskGroup（任务组）与参数化模板，避免复制粘贴整套 DAG。

from datetime import datetime from airflow import DAGfrom airflow.operators.python import PythonOperator def extract():    print("从源系统抽取数据") def transform():    print("清洗与转换数据") def load():    print("写入目标数据仓库") with DAG(    dag_id="etl_pipeline",    start_date=datetime(2025, 1, 1),    schedule="@daily",    catchup=False,    tags=["etl", "example"],) as dag:    t_extract = PythonOperator(task_id="extract", python_callable=extract)    t_transform = PythonOperator(task_id="transform", python_callable=transform)    t_load = PythonOperator(task_id="load", python_callable=load)     t_extract >> t_transform >> t_load

核心能力详解：4 大支柱

支柱一：DAG 编程模型。 DAG（Directed Acyclic Graph，有向无环图）指节点是任务、边是依赖、数据只能向前流动且不允许成环的图结构。Airflow 用普通 Python 文件描述这种结构，因此 DAG 可以进 Git 做版本管理、可以在 CI 里做单测、可以复用公共函数——这是 cron 脚本完全不具备的可维护性。

支柱二：Scheduler 与 Executor 分层调度架构。 Scheduler（调度器）不执行任务，只负责遍历 DAG、判断依赖与时间条件、把任务实例置为可运行状态；Executor（执行器）决定任务“在哪里跑”：LocalExecutor 在本机进程跑，CeleryExecutor 分发到分布式 Worker 队列，KubernetesExecutor 为每个任务拉起独立 Pod。分层设计让“调度逻辑”和“执行资源”解耦，同一套 DAG 可以在单机到集群之间平滑迁移。

支柱三：Operator / Sensor / Hook 三件套与 Provider 生态。 Operator（操作符）是任务的模板，如 PythonOperator 执行函数、BashOperator 执行命令、SnowflakeOperator 执行 SQL；Sensor（传感器）是等待外部条件的特殊任务，如等待文件到达、等待 SQL 查询有返回；Hook（连接钩子）封装外部系统的连接管理，供 Operator 复用。官方与社区维护 80+ 个 Provider 包（AWS、GCP、Spark、Snowflake、Flink 等），基本覆盖主流数据系统，接入新系统通常不需要自己写客户端代码。

支柱四：数据感知调度与可观测性。 除了 cron 时间触发，Airflow 2.4+ 支持 Datasets（数据集）触发：上游任务产出数据集后，依赖该数据集的下游 DAG 自动运行，实现“数据就绪才跑”而非“到点就跑”。可观测性方面，内置 Web UI 提供 DAG 列表、运行历史、依赖图、Gantt 图与任务日志；对外提供稳定的 REST API（/api/v1）和 statsd 指标输出，便于接入企业监控体系。

生产部署与监控：从上线到运维

部署方式一：Docker Compose 官方编排。 官方提供生产可用的 docker-compose 文件，包含 PostgreSQL（元数据库）、Redis（Celery Broker）、webserver、scheduler、worker、triggerer 等 8 个服务，适合中小团队快速搭建标准环境。先下载与版本匹配的 docker-compose.yaml，创建 ./dags ./logs ./plugins ./config 目录并写入 AIRFLOW_UID=$(id -u) 到 .env，然后 docker compose up -d，初始化完成后访问 8080 端口即可。这是官方推荐的本地生产模拟环境。

部署方式二：Celery / Kubernetes 分布式集群。 生产主力方案是 CeleryExecutor：PostgreSQL 存元数据、Redis/RabbitMQ 做消息队列、多台 Worker 机消费任务，2.2+ 可同时运行多个 Scheduler 实现高可用（多个调度器实例通过元数据库协调，避免重复调度）。需要弹性伸缩时改用 KubernetesExecutor，按任务粒度拉起 Pod。关键运维点是 DAG 文件必须同步到所有 scheduler/webserver 节点（共享存储或打入镜像），否则会出现“任务在执行却找不到 DAG 定义”的问题。

可观测性。 三件套组合：任务运行状态看 Airflow UI（Grid 视图 + SLA 统计）；Celery 队列状态用 Flower 监控；指标接入 Prometheus/Grafana 需要配置 statsd 输出（官方 Helm Chart 提供 statsd-exporter 选项）。日志建议统一收集到 ELK/Loki，便于跨节点检索；失败告警可用内置邮件通知，或通过自定义 on_failure_callback 接入钉钉/企业微信/Slack。

<p class="para-label">避坑清单：</p>

<ul class="dash-list"><li>包名陷阱：PyPI 上存在名为 airflow 的第三方包，安装它不会得到官方 Airflow；正确命令是 pip install apache-airflow。</li><li>时区与 start_date：默认采用 UTC，start_date 设为未来日期会导致 DAG 一直不触发；生产应显式配置 default_timezone，并按业务时区核对调度时间。</li><li>多节点文件同步：DAG 文件必须同步到所有 scheduler/webserver 节点（共享存储或镜像打包），否则任务执行时报 DAG missing。</li><li>大数据慎用 XCom：默认 XCom 存数据库，传大对象会撑爆元数据库；大数据应写入对象存储并在 XCom 中传路径。</li><li>安装/升级不用 constraints：Airflow 依赖面大，直接 pip install 极易版本冲突，官方强烈建议用约束文件锁定版本。</li></ul>

版本迭代

Airflow 从 2.0 起全面重写调度器，主线是“稳定性优先 + 渐进式现代化”。2.6 引入多时区调度，2.7 将多 Scheduler 高可用变为默认行为，2.8–2.9 强化数据感知调度（Datasets 与 Object Storage），2.10 将新版 UI 设为默认并引入 DAG 版本化，整体向声明式与云原生演进，为 3.0 铺路（截至训练数据，可能已有更新版本）。

AI 应用场景

常见误区

面试题

参考来源

• webpageApache Airflow 官网项目首页、教程与版本发布入口