Apache Airflow指南：47k星工作流编排平台DAG编写与部署实践

项目：apache/airflow | 星数：47308 | 语言：Python | 协议：Apache-2.0 | GitHub仓库

更新日期：2026-10-29 · 作者：CodeArts 编辑部

什么时候真的需要 Airflow

如果一个任务每天定时跑一次、失败了手动重跑，cron 足够。当出现这些信号时，才值得上 Airflow：任务之间有依赖（B 必须在 A 成功后跑）、需要看每个环节的状态和日志、某一步失败要自动重试、历史数据要补跑、多个团队共享调度能力。

Airflow 的本质是把工作流当成代码：用 Python 写 DAG，任务依赖是代码里的关系，调度器读代码执行。47308 星、Apache-2.0、社区极活跃，是数据工程领域的事实标准。

核心概念：五个名词就够

第一个 DAG：三步 ETL

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

def extract(): ...
def transform(): ...
def load(): ...

with DAG(
    dag_id='daily_etl',
    start_date=datetime(2026, 10, 1),
    schedule='0 2 * * *',
    catchup=False,
    default_args={'retries': 2, 'retry_delay': timedelta(minutes=5)},
    tags=['etl'],
) 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

两个参数值得特别说明。catchup=False 表示不为历史日期补跑——如果 start_date 设成两年前又忘了这个参数，Airflow 会一次性排队两年份的任务，这是新手最常见的"上线即雪崩"事故。

用 TaskFlow API 写更清爽

from airflow.decorators import dag, task

@dag(start_date=datetime(2026, 10, 1), schedule='0 2 * * *', catchup=False)
def daily_etl():
    @task
    def extract() -> list:
        return [1, 2, 3]

    @task
    def transform(rows: list) -> int:
        return sum(rows)

    transform(extract())

daily_etl()

函数返回值自动通过 XCom 传给下游，不用手写 ti.xcom_push。注意 XCom 存在元数据库里，只适合传小数据（ID、几十 KB 的 JSON），几百 MB 的数据应落到对象存储再传路径。

部署方式怎么选

最快的上手方式：

curl -LfO https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml
mkdir -p ./dags ./logs ./plugins
echo -e "AIRFLOW_UID=$(id -u)" > .env
docker compose up airflow-init
docker compose up -d

起来后访问 8080 端口，默认账号密码都是 airflow。SQLite 是默认元数据库，只适合单机体验，多 Worker 必须换成 Postgres。

生产环境的四个关键配置

• Executor 选择：小规模用 LocalExecutor（单机多进程），生产用 CeleryExecutor 或 KubernetesExecutor。
• DAG 解析开销：所有 DAG 文件会被反复导入解析，顶层别写网络请求和数据库查询，否则调度器会被拖垮。用 schedule=None 加手动触发可以避免无意解析。
• 池（Pool）限流：给访问同一数据库的任务分配 pool，限制并发槽位，防止把业务库打挂。
• 日志清理：元数据库增长很快，配上 airflow db clean 定期任务，否则几个月后查询页面会明显变慢。

回填与重跑的正确姿势

airflow dags backfill -s 2026-10-01 -e 2026-10-07 daily_etl
airflow tasks test daily_etl extract 2026-10-29
airflow tasks clear daily_etl -s 2026-10-20 -e 2026-10-25 -t load

tasks test 在本地直接跑单个任务不写元数据库，调试很好用；backfill 补历史区间；clear 把指定时间段的实例重置待重跑。补数据前一定确认任务是否幂等，否则重复写入会产生脏数据。

和替代品的边界

Airflow 适合批量、定时、有依赖的数据管道。如果是事件驱动的毫秒级触发、复杂分支重试，Temporal、Prefect、Dagster 可能更合适；如果是简单几个 shell 脚本串起来，cron + 一个 shell 脚本就够了。选型的核心不是谁的星多，而是你的工作流是什么形状。