dagster

Dagster 数据资产编排

补充说明

Dagster 以“数据资产（asset）”为中心，适合管理数据管道的依赖、分区、调度和血缘。核心概念是 asset、job、schedule、sensor 和 Definitions。

常用示例

本地开发：

dagster dev -f definitions.py
# 访问 http://localhost:3000

定义资产：

from dagster import asset, Definitions, ScheduleDefinition

@asset
def raw_orders():
    return [{"id": 1, "amount": 100}]

@asset
def daily_revenue(raw_orders):
    return sum(o["amount"] for o in raw_orders)

defs = Definitions(
    assets=[raw_orders, daily_revenue],
    schedules=[ScheduleDefinition(
        job=...,
        cron_schedule="0 2 * * *",
    )],
)

执行资产：

dagster asset materialize -f definitions.py --select daily_revenue

查看运行记录：

dagster run list
dagster asset list

生产部署使用 dagster-webserver、dagster-daemon 和 PostgreSQL，并配置对象存储保存运行日志。

易错点与排错

• 资产函数返回值没有类型或序列化失败，I/O manager 报错。
• 分区定义和调度不匹配，历史分区没有回填。
• Definitions 没有注册 job/schedule，UI 看不到。
• daemon 没有运行，定时调度和传感器不触发。
• 生产使用本地文件存储日志，容器重建后丢失。
• 资产依赖写错，血缘图与实际数据不一致。
• 依赖包未锁定，部署环境与开发环境行为不同。
• 运行并发过高，数据库连接池耗尽。