构建 Dagster 资产 是做什么的？
使用时机
- 创建或重构 Dagster 软件定义资产和作业。
- 将表/文件/ML模型建模为带有血缘关系的资产。
- 添加分区、回填、资产检查、调度或传感器。
- 不要用于 Airflow（请使用 Airflow 相关技能）。
工作流
- [ ] 将每个输出建模为 @asset；通过函数参数声明依赖
- [ ] 为时间/类别切片的数据添加分区
- [ ] 将 IO（读/写）移入 IO 管理器或资源
- [ ] 添加资产检查以保障数据质量
- [ ] 设置调度/传感器来自动物化
- 以资产而非任务来思考。 资产是一个持久化对象（表、文件、模型）。通过将上游资产作为函数参数来声明依赖关系——Dagster 会自动构建血缘图。
- 对自然切片的数据（按天、区域）进行分区，以便能够一次物化/回填一个切片。
- 资源和 IO 管理器持有连接和读/写逻辑，使资产主体专注于转换，并使其可测试。
- 资产检查将数据质量断言附加到资产上。
模式
带有依赖关系的分区资产：
import dagster as dg
daily = dg.DailyPartitionsDefinition(start_date="2026-01-01")
@dg.asset(partitions_def=daily)
def raw_orders(context: dg.AssetExecutionContext) -> None:
day = context.partition_key
write_parquet(f"raw/orders/{day}.parquet", fetch_orders(day))
@dg.asset(partitions_def=daily)
def orders_clean(context, raw_orders) -> None: # 依赖于 raw_orders
day = context.partition_key
transform_and_load(day)
@dg.asset_check(asset=orders_clean)
def no_null_ids(context) -> dg.AssetCheckResult:
n = count_null_order_ids(context.partition_key)
return dg.AssetCheckResult(passed=n == 0, metadata={"null_ids": n})
资源 / IO 管理器 — 注入一个仓库客户端，而不是在资产内部构造它：
@dg.asset
def orders_summary(context, warehouse: WarehouseResource, orders_clean):
warehouse.execute("insert into summary select ...")
调度一个分区作业，使得每次运行都能物化最新的分区；使用传感器在上游文件/资产出现时触发物化。
常见陷阱
- 任务思维 — 对所有操作都使用裸
@op/jobs 会失去血缘关系、可观测性以及资产免费提供的分区感知回填功能。 - 在资产体内编写 IO — 硬编码连接会使资产无法测试；应使用资源/IO 管理器。
- 非幂等的分区资产 — 物化一个分区必须能够覆盖该分区，而非追加（参见
writing-idempotent-transformations）。 - 跳过资产检查 — 没有检查，不良数据会静默物化；Dagster 会在 UI 中显示检查失败并可阻止下游操作。
- 依赖资产之间分区定义不匹配 — 保持
partitions_def一致，以便解析映射关系。