Dagster数据血缘追踪与资产治理 - 第3节

Dagster 是什么？一个数据工程师的“操作系统”

说实话，我第一次接触 Dagster 的时候，心里想的是：“又一个调度框架？” 那时候我已经被 Airflow 折磨了大半年，Prefect 也试过一阵子。但真正用起来才发现，Dagster 的野心不止是做调度——它想成为数据工程的“操作系统”。

怎么理解呢？你想想看，传统的数据 pipeline，我们关注的是“什么时候跑”、“跑完没”。但 Dagster 更关心的是：“你生产了什么数据？”、“这个数据质量行不行？”、“上下游依赖清不清楚？”。说白了，它把数据本身当成了核心资产，而不是把任务执行当成核心。

我个人习惯把 Dagster 比作“乐高积木 + 说明书”。Ops 是积木块，Assets 是搭好的模型，Jobs 是拼装步骤，Graph 是设计图纸。这套思维模型，让我在项目中少踩了很多坑。

Dagster vs Airflow vs Prefect：我选谁？

这个问题几乎每场分享都会被问到。我直接说结论：没有银弹，但有更适合的场景。下面这张表是我在实际项目中的体感对比。

维度	Airflow	Prefect	Dagster
核心抽象	DAG（有向无环图）	Flow / Task	Ops / Assets / Jobs / Graph
数据意识	弱（不感知数据产出）	中（有 Result 概念）	强（Asset 是一等公民）
本地开发	差（需要模拟调度器）	好（本地运行友好）	极好（Dagster Dev 一键启动）
测试能力	弱（依赖 mock）	中（有测试工具）	强（内置测试框架）
血缘追踪	需插件（如 Lineage）	需额外配置	原生支持（Asset 自动追踪）
学习曲线	中（概念少但坑多）	低（API 简洁）	中高（概念多但逻辑自洽）
社区生态	最成熟	快速增长	快速发展中

Dagster 四大核心概念

嗯，这里要仔细讲。这四个概念搞懂了，Dagster 就算入门了。我当年花了整整一周才理清楚它们之间的关系，其实没那么复杂。

1. Ops：最小的计算单元

Ops 就是一段可复用的计算逻辑。你可以把它理解成一个函数，但它比函数多了输入输出声明和上下文管理。

from dagster import op

@op
def extract_data(context):
    """从 API 拉取原始数据"""
    context.log.info("开始提取数据...")
    # 模拟数据提取
    return [{"id": 1, "value": 100}, {"id": 2, "value": 200}]

@op
def transform_data(context, raw_data):
    """数据清洗和转换"""
    context.log.info(f"处理 {len(raw_data)} 条记录")
    return [{"id": d["id"], "value": d["value"] * 2} for d in raw_data]

注意看，每个 Op 都可以接收 context 参数，这里面有日志、资源配置、分区信息等。我曾经在项目中犯过一个错误：把所有逻辑塞进一个 Op 里，结果调试起来痛苦得要命。记住，一个 Op 只做一件事。

2. Assets：数据资产，Dagster 的灵魂

这是 Dagster 和其他框架最大的区别。Asset 代表一个数据产物——可以是一张表、一个文件、一个 ML 模型。你不再说“运行这个任务”，而是说“产出这个数据资产”。

from dagster import asset

@asset
def cleaned_sales_data(raw_sales_data):
    """清洗后的销售数据表"""
    # raw_sales_data 是上游 Asset 自动注入的
    cleaned = [r for r in raw_sales_data if r["value"] > 0]
    return cleaned

@asset
def daily_report(cleaned_sales_data):
    """每日销售报表"""
    total = sum(r["value"] for r in cleaned_sales_data)
    return {"total_sales": total, "record_count": len(cleaned_sales_data)}

3. Jobs：把 Ops 和 Assets 串起来

Job 就是一组 Ops 或 Assets 的执行计划。你可以把它理解成“一个完整的 pipeline 实例”。

from dagster import define_asset_job

# 基于 Asset 定义 Job
sales_job = define_asset_job(
    name="sales_pipeline",
    selection=["raw_sales_data", "cleaned_sales_data", "daily_report"]
)

# 或者基于 Ops 定义 Job（传统方式）
from dagster import job

@job
def traditional_job():
    result = extract_data()
    transform_data(result)

我个人更推荐用 Asset 方式定义 Job，因为血缘关系一目了然。传统 Op 方式适合那些没有明确数据产出的“副作用”任务，比如发送告警邮件。

4. Graph：Job 的蓝图

Graph 是 Ops 之间的连接拓扑图。你可以把它理解成“可复用的子图”。当你的 pipeline 变得复杂时，Graph 能帮你组织代码。

from dagster import graph

@graph
def etl_pipeline():
    """ETL 子图，可被多个 Job 复用"""
    raw = extract_data()
    cleaned = transform_data(raw)
    return cleaned

# 在 Job 中引用 Graph
@job
def daily_etl_job():
    etl_pipeline()

@job
def weekly_etl_job():
    # 可以添加额外的处理步骤
    result = etl_pipeline()
    load_to_warehouse(result)

核心概念关系图

下面这张图是我自己总结的，每次给团队培训都用它。你看完应该能明白这四个概念是怎么配合的。

小结：为什么我推荐你从 Assets 开始？

如果你刚接触 Dagster，我的建议是：先忘掉 Ops 和 Graph，直接从 @asset 开始写。为什么？因为 Assets 的思维方式最贴近数据工程师的日常——你每天关心的不就是“这张表什么时候更新”、“那个报表数据对不对”吗？

等你用 Assets 把 pipeline 搭起来，再回头看 Ops 和 Graph，你会发现它们只是更底层的工具。就像学开车，先学会踩油门打方向盘，再研究发动机原理，这样更顺。