核心概念（上）：理解 Asset、Op、Graph、Job

好，咱们正式开始啃 Dagster 的核心概念。这一章我带你过四个最基础的东西：Asset（资产）、Op（操作）、Graph（图）、Job（作业）。

这四个概念，说白了就是 Dagster 世界的「砖块」和「图纸」。你搞懂了它们，后面写管道就跟搭积木一样顺手。

我个人习惯把 Asset 放在第一个讲，因为它最贴近业务直觉。咱们先看 Asset。

Asset（资产）：数据管道里的「交付物」

Asset 是什么？就是你管道最终要产出的东西。一张报表、一个清洗后的数据集、一个机器学习模型，都算 Asset。

我刚开始用 Dagster 时，总把它当成「表名」或者「文件路径」来理解。后来发现不对——Asset 其实是一个有元数据的产出物。它不光告诉你「我是什么」，还告诉你「我怎么来的」「我依赖谁」。

核心要点：Asset 是 Dagster 里的一等公民。你定义管道时，先想清楚要产出哪些 Asset，再想怎么算出来。

看个例子：

from dagster import asset

@asset
def raw_sales_data() -> list:
    """从原始数据源拉取的销售记录"""
    return [{"date": "2024-01-01", "amount": 100}]

@asset
def cleaned_sales_data(raw_sales_data: list) -> list:
    """清洗后的销售数据"""
    return [r for r in raw_sales_data if r["amount"] > 0]

你看，cleaned_sales_data 自动依赖 raw_sales_data。Dagster 会帮你追踪这个依赖关系。我在项目中遇到过有人手动维护依赖表，结果一改字段就崩——用 Asset 注解后，依赖关系自动更新，省心多了。

我的习惯：每个 Asset 只做一件事。比如「清洗」和「聚合」拆成两个 Asset，而不是揉在一起。这样调试时能精准定位问题。

Op（操作）：最小的执行单元

Op 是 Dagster 里最基础的「动作」单元。它接收输入，产生输出，中间干点活。

你可以把 Op 想象成一个函数——但它比函数多了两个东西：明确的输入输出声明和可配置的参数。

from dagster import op, Out, In

@op(
    ins={"raw_data": In(dagster_type=list)},
    out={"cleaned": Out(dagster_type=list)}
)
def clean_data(raw_data: list) -> list:
    """清洗原始数据"""
    return [item for item in raw_data if item["status"] == "active"]

嗯，这里要注意：Op 的输入输出类型声明不是摆设。Dagster 会在运行时做类型检查。我曾经因为没写类型，结果一个字符串混进了整数列表，跑了两小时才发现——从那以后，我每个 Op 都老老实实写类型。

Op 和 Asset 有什么区别？简单说：

  • Asset 关注「产出什么」
  • Op 关注「怎么产出」

你完全可以只用 Op 搭管道，但加上 Asset 后，整个管道的可观测性会提升一个档次。

Graph（图）：把 Op 串起来

单个 Op 能干的事有限。把多个 Op 连起来，就形成了 Graph。

Graph 定义了 Op 之间的数据流向。谁先执行，谁后执行，数据怎么传递——都在 Graph 里说清楚。

from dagster import graph

@graph
def my_pipeline():
    raw = fetch_data()
    cleaned = clean_data(raw)
    result = aggregate_data(cleaned)
    return result

你想想看，这个 my_pipeline 就是一个 Graph。它把三个 Op 串成一条线。Dagster 会自动解析依赖，决定执行顺序。

我见过有人把几十个 Op 塞进一个 Graph，结果调试时根本看不清谁依赖谁。我的建议是：一个 Graph 控制在 5-8 个 Op 以内。超过这个数，就该拆成子 Graph 了。

避坑指南：Graph 里不要写业务逻辑。它只负责编排，不负责计算。我曾经犯过这个错——在 Graph 里加了个 if-else，结果测试时逻辑混乱，改了半天。业务逻辑请放在 Op 里。

Job（作业）：可执行的管道

Graph 是「图纸」，Job 是「开工指令」。你把 Graph 加上运行配置（比如调度时间、资源限制、环境变量），就变成了 Job。

from dagster import job, Config

class MyConfig(Config):
    batch_size: int = 100
    retry_count: int = 3

@job(config=MyConfig)
def daily_sales_job():
    my_pipeline()

Job 是 Dagster 里真正能跑的东西。你可以通过 Dagster UI 手动触发，也可以配成定时任务。

我个人习惯把 Job 按「触发方式」命名：daily_sales_job、hourly_clean_job、manual_retry_job。这样一看名字就知道它什么时候跑。

四者关系：一张图说清楚

下面这张 SVG 图展示了 Asset、Op、Graph、Job 之间的关系：

Dagster 核心概念关系图 Asset（资产） 定义「产出什么」—— 数据表、模型、报表 Op（操作） 定义「怎么产出」—— 最小的执行单元 Graph（图） 定义「执行顺序」—— 把 Op 串成有向无环图 Job（作业） 定义「何时运行」—— Graph + 运行配置 Asset 定义「做什么」，Op 定义「怎么做」，Graph 定义「顺序」，Job 定义「何时跑」 四者层层封装，从业务层到执行层逐步具体化

对比总结

这四个概念容易混淆，我整理了一张对比表：

概念 关注点 是否可执行 依赖关系 典型用途
Asset 产出物 否 自动追踪 定义数据产品
Op 计算逻辑 否 手动声明 封装业务逻辑
Graph 执行顺序 否 编排关系 组合多个 Op
Job 运行配置 是 无 调度执行

我的建议：刚开始学 Dagster，先只用 Asset 和 Job。等熟悉了再引入 Op 和 Graph 做更精细的控制。别一上来就搞四件套，容易晕。

好了，这一章的核心概念就这些。Asset 是「做什么」，Op 是「怎么做」，Graph 是「顺序」，Job 是「时机」。你把这四个概念刻在脑子里，后面学调度、资源、传感器时就会轻松很多。

← 上一章 📖 返回目录 下一章 →