Airflow的XCom传了个10GB的DataFrame——任务间数据传递

# 正确用法：XCom传路径，数据走S3
 
# 上游Task：计算结果写S3，XCom传路径
@task
def transform_data(ds):
    df = spark.read.parquet(f's3://data/dwd/dt={ds}')
    result = df.groupBy('country').agg(sum('amount'))
    output_path = f's3://data/dws/dt={ds}'
    result.write.parquet(output_path)
    return output_path  # TaskFlow自动通过XCom传递
 
# 下游Task：从XCom拿路径，从S3读数据
@task
def export_report(output_path):
    df = spark.read.parquet(output_path)  # 从S3读
    push_to_bi(df)
 
# DAG中
path = transform_data('{{ ds }}')
export_report(path)
 
# XCom只传了一个字符串路径（几十字节）
# 实际数据在S3上（可以是TB级）

Airflow Variable vs XCom

维度	XCom	Variable
存储	MetaDB（和Task Run绑定）	MetaDB（全局）
生命周期	随DAG Run创建和清理	手动管理（持久存在）
用途	任务间传递运行时数据	存储全局配置
示例	上游产出的文件路径	数据库连接信息、API Key
大小限制	几KB到几MB（看DB配置）	几KB

实战踩坑：XCom撑爆MetaDB

在某知名出海硬件品牌，有个同事在Airflow的@task函数中直接return了一个Pandas DataFrame（约500MB）。TaskFlow API会自动把返回值序列化到XCom（存在PostgreSQL的xcom表里）。

结果：PostgreSQL的xcom表暴增到几十GB，数据库查询变慢，Airflow Web UI打不开，Scheduler扫描DAG变慢。最终不得不手动TRUNCATE xcom表才恢复。

修复：① return改成返回S3路径而不是DataFrame本身；② 在airflow.cfg中设xcom_backend使用自定义的S3 XCom Backend（大数据自动存S3只在XCom表存引用）；③ 设定期清理XCom的DAG——每周清理30天前的XCom记录。

面试高频追问

追问一：TaskFlow API的返回值都走XCom吗？

是的。@task装饰的函数的返回值自动序列化到XCom。所以一定注意返回值的大小——返回路径/ID/状态等小数据，不要返回DataFrame/大列表/大字典。如果需要传大数据就手动写S3后返回路径。

追问二：XCom Backend是什么？

Airflow 2.0+支持自定义XCom Backend——把XCom数据存到S3/GCS而不是数据库。大的XCom值自动上传到对象存储，数据库只存引用。适合偶尔需要传较大数据（几MB）的场景。但最佳实践还是控制XCom大小不依赖Backend。

追问三：DolphinScheduler怎么传递任务间数据？

DS没有XCom概念。任务间数据传递主要靠：① 全局变量/本地变量——在工作流中定义变量上下游共享；② 中间表/文件——和Airflow一样上游写下游读。DS的方式更简单但灵活度不如XCom。

总结

任务间数据传递讲两个原则：① XCom只传小数据（路径/ID/状态/配置，几KB以内），大数据走中间存储（S3/Hive表）；② TaskFlow的返回值=XCom，控制返回值大小。能讲出XCom撑爆MetaDB的踩坑和改S3路径传递的修复方案，面试官觉得你理解了调度系统的设计约束。

下一篇聊触发方式设计：定时任务和事件驱动怎么混合使用？

你们任务间怎么传数据？踩过XCom的坑吗？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长