Airflow 做 ETL，真不是“排个 DAG 就完事儿”：那些年我踩过的坑与悟出的道

大家好，我是 Echo_Wish，一个在大数据 ETL 世界里摸爬滚打多年、见过无数 Airflow“惨案”的人。

很多同学以为 Airflow = 画个 DAG + 写个 Operator + 放到生产跑，图简单、逻辑清晰、界面也漂亮，看起来“稳得一批”。

但真上生产后你会发现：
Airflow 好用，但绝对不会自动变好用。
配置不合理、调度不规范、任务不隔离、Operator 滥用、XCom 滥发消息……这些问题不踩上几次，根本意识不到它们有多“致命”。

今天我就跟你聊聊：
怎么用 Airflow 搭一个真的能扛生产的 ETL 系统 —— 并顺便告诉你哪些坑必须绕开。

一、Airflow 最容易犯的错误：把它当“任务执行器”而不是“调度编排器”

我见过不少项目把 Airflow 当成“万能胶”：

• 数据清洗写在 PythonOperator
• 数据加工写在 BashOperator
• 数据入仓也写在 PythonOperator
• 拖个 DockerOperator 跑 Spark 任务
• 甚至有团队把 MySQL 的小 SQL 都塞到 PythonOperator 里执行…

结果 DAG 看起来像一大坨年糕，谁也不敢动，改一个节点要跪十分钟祈祷不爆炸。

正确做法应该是：

✔ Airflow 做 编排 —— 决定什么任务什么时候跑
✔ 真正的数据处理任务交给 Spark / Flink / 任务脚本
✔ Airflow 只负责触发 & 监控，不负责计算

所以 在 DAG 里偷懒写大量逻辑，是最不应该的行为。

二、上生产前必须具备的 ETL DAG 基本形态

话不多说，给你一个能上生产的基础 DAG 模版，里面包含了一些最佳实践。

示范代码：带重试、隔离、任务边界清晰的 ETL DAG

from datetime import datetime, timedelta
from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from airflow.operators.empty import EmptyOperator

default_args = {
    "owner": "echo_wish",
    "email": ["alert@datapipeline.com"],
    "email_on_failure": True,
    "retries": 2,                # 必须要有重试
    "retry_delay": timedelta(minutes=5),
    "depends_on_past": False,    # ETL 不推荐依赖过去状态
}

with DAG(
    dag_id="etl_sales_daily",
    start_date=datetime(2024, 1, 1),
    schedule_interval="0 1 * * *",    # 每天凌晨1点跑
    catchup=False,                    # 避免一次补几千天的“悲剧”
    default_args=default_args
) as dag:

    start = EmptyOperator(task_id="start")

    extract = KubernetesPodOperator(
        task_id="extract_raw_data",
        name="extract-task",
        namespace="airflow-jobs",
        image="etl/extract:latest",
        cmds=["python", "extract.py"],
        arguments=["--date", "{{ ds }}"],
        is_delete_operator_pod=True
    )

    transform = KubernetesPodOperator(
        task_id="transform_data",
        name="transform-task",
        namespace="airflow-jobs",
        image="etl/transform:latest",
        cmds=["python", "transform.py"],
        arguments=["--date", "{{ ds }}"],
        is_delete_operator_pod=True
    )

    load = KubernetesPodOperator(
        task_id="load_to_dwh",
        name="load-task",
        namespace="airflow-jobs",
        image="etl/load:latest",
        cmds=["python", "load.py"],
        arguments=["--date", "{{ ds }}"],
        is_delete_operator_pod=True
    )

    end = EmptyOperator(task_id="end")

    start >> extract >> transform >> load >> end

你会发现几个核心点：

• 用 KubernetesPodOperator 跑任务，真正做到完全隔离
• 每一步都拆开，逻辑清晰明了
• ETL 代码不写在 DAG 里，Airflow 只是个 orchestrator
• 重试、邮件、调度、任务边界全部规范

如果你现在的 DAG 长得不像这样，那大概率有优化空间。

三、Airflow 搭 ETL 经常被忽视的关键设计

1. DAG 要小，不要大

很多项目喜欢弄一个“大而全”的 DAG，每天跑几十个节点。
出了问题根本不知道是哪个子流程挂了。

更优解：

✔ 一条业务链路一个 DAG
✔ 一个表一个 DAG（特别是快照/宽表）
✔ 公共依赖拆成子 DAG 或者单独维护

越小越好管理，这是真理。

2. XCom 慎用：不要把大对象丢进去

我见过最魔幻的 Airflow 事故：

某同事把一个 100MB 的 Pandas DataFrame 通过 XCom 往下游传……

Airflow 的 metadata DB（MySQL/Postgres）瞬间爆炸。

原则：XCom 只能传 Metadata、小量字符串，不传数据本体。

怎么传数据？
✔ 上传到 OSS/S3/HDFS
✔ XCom 里只放路径

3. 不要把 Airflow 当成“查询引擎”

反模式例子：

# 千万不要这样写
def really_bad_task():
    import pandas as pd
    df = pd.read_sql("SELECT * FROM big_table", conn)
    ... # 本地处理

你会让 scheduler / worker 直接打 DB，造成：

• DB 压力飙升
• Worker 内存打爆
• 性能惨不忍睹

正确姿势：

✔ DB → 外部脚本/Spark 处理
✔ Airflow 只触发脚本，不做 heavy load

4. 所有 ETL 镜像必须可重复执行、幂等、安全回滚

为什么？

因为 Airflow 的哲学是：

任务可以失败，但不能留下脏数据。

所以你的 ETL 脚本必须遵守：

• 同一天跑多少次，结果必须一致
• 如果失败要能回滚
• load 阶段要么全成功，要么全部失败

四、我踩过最痛的坑：依赖管理混乱

Airflow 里依赖一旦搞乱，你会看到：

• 任务 A 明明成功了，但任务 B 不跑
• 因为一个节点卡住，整个 DAG 挂在那里十小时
• 新增节点导致循环依赖，直接跑不起来

最糟糕的“反面教材”就是写成树状 + 多个分支互相引用。

正确思路：它必须是一条链，或者清晰的树，而不是蜘蛛网。

如果 DAG 长这样，那你就成功打造了生产事故：

[外链图片转存中…(img-cNI7A7JU-1765548424463)]

五、关于监控：Airflow 不等于运维系统

Airflow 自带的监控并不够，它只告诉你：

• 成功
• 失败
• 超时
• 重试

但真正的生产 ETL 需要：

✔ 数据量监控（比如今天入库 100W，昨天 200W，是不是异常？）
✔ 数据质量监控（空值、重复、业务约束）
✔ 延迟监控（下游 SLA）
✔ 元数据记录（血缘、字段变更）

所以 Airflow 不能独立支撑所有需求，必须补上：

• Prometheus + Grafana
• 数据质量工具（Great Expectations、Deequ）
• 元数据管理（DataHub、Amundsen）

六、写在最后：Airflow 本质上是“纪律工具”

这么多年用下来我有一个特别深的感受：
Airflow 的根问题不在技术，而在团队习惯。

• 如果大家喜欢把逻辑塞到 DAG 里 → DAG 会腐烂
• 如果大家不遵守幂等、分离职责、轻逻辑原则 → ETL 会混乱
• 如果业务不愿意规范数据结构 → 后端永远在救火

Airflow 是个非常成熟、强大、稳定的调度系统。
真正让它变得“不稳定”的，是使用它的人。

只要记住一句话：

Airflow 做 ETL，不是搭一个系统，而是培养一套团队工程文化。