📚 课程目录第4章 DAG定义方式

4、DAG定义方式：Airflow的Python上下文管理器 vs Dagster的@job装饰器与GraphDefinition

聊到DAG的定义方式，其实就是在聊「这个工作流到底怎么描述」。我见过不少团队，代码写得挺溜，但DAG定义那层却搞得一团糟。说白了，定义方式决定了你后续调试、重构、测试的体验。

Airflow和Dagster在这条路上走了完全不同的方向。一个靠上下文管理器隐式收集任务，一个靠装饰器显式组装。嗯，咱们今天就把这事掰扯清楚。

Airflow：Python上下文管理器的「隐式收集」

Airflow的DAG定义，核心就是 with DAG(...) as dag: 这个上下文管理器。你想想看，所有在with块里实例化的Operator，都会被自动「吸」进这个DAG里。

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

def extract():
    print("抽取数据")

def transform():
    print("转换数据")

def load():
    print("加载数据")

with DAG(
    dag_id="etl_pipeline",
    start_date=datetime(2024, 1, 1),
    schedule="@daily",
    catchup=False,
) as dag:

    task_extract = PythonOperator(
        task_id="extract",
        python_callable=extract,
    )

    task_transform = PythonOperator(
        task_id="transform",
        python_callable=transform,
    )

    task_load = PythonOperator(
        task_id="load",
        python_callable=load,
    )

    task_extract >> task_transform >> task_load

这段代码看着挺清爽对吧？但我得提醒你一个坑：依赖关系是靠位移操作符（>>）显式声明的，不是自动推断的。我曾经接手过一个Airflow项目，里面有40多个任务，依赖关系画出来跟蜘蛛网似的。你猜怎么着？有人忘了写某个 >>，结果任务跑完了下游没触发，数据对不上，排查了整整两天。

注意：Airflow的上下文管理器只负责「收集任务」，不负责「推断依赖」。任务之间的上下游关系必须手动用 >> 或 set_downstream() 声明。漏掉一个，DAG就断了。

另外，Airflow的DAG定义其实是个Python脚本，每次调度器解析时都会重新执行一遍。这意味着你在模块顶层写的那些耗时的数据库查询、API调用，都会在解析时重复执行。我见过有人把 pd.read_csv() 直接写在DAG文件里，结果调度器每30秒解析一次，数据库连接池直接被打满。

建议：把重量级初始化逻辑放到Operator的 execute() 方法里，或者用 DAG.run() 的上下文。别让调度器解析阶段干太多活。

Dagster：@job装饰器的「显式组装」

Dagster走了另一条路。它用 @job 装饰器把一组 @op 装饰的函数组装成一个DAG。每个op的输入输出通过参数和返回值显式传递，依赖关系自动推导。

from dagster import job, op

@op
def extract():
    data = [1, 2, 3, 4, 5]
    return data

@op
def transform(data):
    result = [x * 2 for x in data]
    return result

@op
def load(result):
    print(f"加载数据: {result}")

@job
def etl_pipeline():
    load(transform(extract()))

你看，这里没有 >>，没有上下文管理器。依赖关系完全由函数调用链决定。我个人习惯这种方式，因为它更接近普通Python函数的直觉——你传参，它返回，依赖自然就建立了。

但Dagster也提供了更灵活的 GraphDefinition，适合需要动态组装或复用的场景：

from dagster import GraphDefinition, op

@op
def step_a():
    return "A"

@op
def step_b(input_a):
    return f"{input_a} -> B"

@op
def step_c(input_b):
    return f"{input_b} -> C"

# 用 GraphDefinition 显式构建
my_graph = GraphDefinition(
    name="my_pipeline",
    node_defs=[step_a, step_b, step_c],
    dependencies={
        step_b: {"input_a": step_a},
        step_c: {"input_b": step_b},
    },
)

# 注册为 job
my_job = my_graph.to_job()

这里要注意：GraphDefinition 的依赖声明是显式的，但比Airflow的 >> 更结构化——它明确指定了「哪个op的哪个输入来自哪个op的哪个输出」。我遇到过团队用Airflow时，任务间传递数据全靠XCom，参数名对不上就崩。Dagster这种显式绑定，至少编译期就能发现参数不匹配的问题。

核心差异对比

维度	Airflow（上下文管理器）	Dagster（@job + GraphDefinition）
依赖推导方式	手动用 `>>` 声明	自动从函数调用链推导，或显式声明
数据传递	通过 XCom 隐式传递	通过函数参数/返回值显式传递
复用性	低，DAG 文件是整体脚本	高，op 和 GraphDefinition 可独立复用
测试友好度	中等，需要 mock DAG 上下文	高，op 就是普通函数，直接单元测试
动态 DAG	支持，但容易写出「解析时爆炸」的代码	原生支持，通过 `@graph` 和 `DynamicOut`

一张图看懂两种定义哲学

下面这张SVG图，把两种方式的「定义-解析-执行」链路画清楚了。你一看就明白，Airflow的隐式收集和Dagster的显式组装，本质区别在哪。

我的选择建议

如果你团队里都是Python老手，项目规模不大，Airflow的上下文管理器确实上手快。但如果你要构建一个长期维护、多人协作的数据平台，我建议你认真考虑Dagster的显式组装方式。

为什么？因为显式比隐式更容易调试和测试。我在一个金融项目中吃过亏：Airflow的DAG里有个任务依赖写错了，导致生产环境的数据管道跑了三个月才发现少了一个转换步骤。后来切到Dagster，每个op都是独立函数，单元测试覆盖率直接拉到90%，再也没出过这种低级错误。

当然，没有银弹。Airflow的生态更成熟，插件更多；Dagster还在快速迭代中。但单从DAG定义方式这个维度看，Dagster的设计更符合现代软件工程的最佳实践——显式优于隐式，组合优于继承。