TaskFlow 是什么
TaskFlow 是 Airflow 2.0 引入的写法：用 @task 装饰器把普通 Python 函数变成任务，用 @dag 装饰器把函数变成 DAG。任务之间通过调用关系自动建立依赖，返回值自动经 XCom 传给下游，省掉了手写 task_id、传参和 xcom_pull 的样板代码。它适合以 Python 逻辑为主、数据量轻量的编排；重计算仍应交给 Spark 等外部引擎，通过连接器算子去触发。
基本信息
| 装饰器 | @task（任务）与 @dag（DAG） |
|---|---|
| 数据传递 | 函数返回值自动 XCom 序列化 |
| 建立依赖 | 任务函数调用任务函数即可 |
| 版本要求 | Airflow 2.0 及以上 |
| 适用边界 | 纯 Python 逻辑、计算轻量的编排 |
| 与算子关系 | 可与传统 Operator 混用，需显式声明依赖 |
TaskFlow 写法
TaskFlow 写法
from datetime import datetime
from airflow.decorators import dag, task
@dag(start_date=datetime(2026, 1, 1), schedule="0 3 * * *", catchup=False, tags=["etl"])
def etl_demo():
@task
def extract() -> dict: # 返回值自动进 XCom
return {"rows": 100, "path": "s3://lake/dwd/orders/dt=2026-10-11"}
@task
def load(meta: dict) -> None: # 上游返回值直接作为参数传入
print(meta["rows"], meta["path"])
load(extract()) # 调用即声明依赖
etl_demo()典型使用场景
- 纯 Python 的 ETL 小任务：清洗、校验、通知这类不需要外部引擎的步骤
- 需要动态任务图：按上游返回的列表动态展开一批下游任务（动态任务映射）
- 参数化执行：通过 dag_run.conf 或 dag_run 上下文取运行时参数
- 替代传统的 PythonOperator：代码更短、类型标注清晰、依赖一眼可见
- 与连接器算子混用：TaskFlow 函数负责轻处理，提交重活仍用专用算子
- 封装可复用逻辑：把公共步骤抽成普通函数再在多个 @task 里调用
常见用法读法
| 写法 | 作用 | 注意 |
|---|---|---|
| @task 修饰函数 | 把函数注册成任务 | 函数名即 task_id，改名会影响历史 |
| 直接调用任务函数 | 声明依赖并传参 | 不要用 xcom_pull 绕一圈 |
| 任务的返回值 | 自动序列化进 XCom | 只放小体积元数据 |
| with TaskGroup(...) | 对任务分组 | 旧版可用 SubDagOperator，已不推荐 |
| 上下文变量 | 取 ds、dag_run、task 等 | 通过参数注入而不是全局取 |
| 动态任务映射 | 按列表批量展开下游 | 展开的数量不宜过大，会放大调度压力 |
XCom 适合传小数据（行数、路径、元信息）；几万行结果别塞 XCom，会写进元数据库拖垮调度。
TaskFlow 函数在 worker 上执行，函数体内不要依赖调度器侧的全局变量；跨任务共享的配置应通过 Airflow 变量、连接或参数传入，这样本地测试与线上行为才一致。