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

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

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

但真上生产后你会发现：

Airflow 好用，但绝对不会自动变好用。

配置不合理、调度不规范、任务不隔离、Operator 滥用、XCom 滥发消息……这些问题不踩上几次，根本意识不到它们有多“致命”。

今天我就跟你聊聊：

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

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

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

结果 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

你会发现几个核心点：

如果你现在的 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 → 外部脚本/Spark 处理

Airflow 只触发脚本，不做 heavy load

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

为什么？

因为 Airflow 的哲学是：

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

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

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

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

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

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

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

外链图片转存中...(img-cNI7A7JU-1765548424463)

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

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

但真正的生产 ETL 需要：

数据量监控（比如今天入库 100W，昨天 200W，是不是异常？）

数据质量监控（空值、重复、业务约束）

延迟监控（下游 SLA）

元数据记录（血缘、字段变更）

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

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

这么多年用下来我有一个特别深的感受：

Airflow 的根问题不在技术，而在团队习惯。

Airflow 是个非常成熟、强大、稳定的调度系统。

真正让它变得“不稳定”的，是使用它的人。

只要记住一句话：

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