Dagster：云原生数据管道编排，从开发到生产全包了

Dagster 在 GitHub 上已经拿到 15,647 Star 了。

做数据工程的人都清楚 ETL 管线的维护成本。开发环境能跑的数据流一上生产就崩，数据血缘全靠人工记，出了故障要在多个系统之间来回翻日志。传统调度工具在规模和可测试性上总有短板。

Dagster 是一个云原生数据管道编排器，覆盖从本地开发到生产部署的完整生命周期。由总部位于纽约的 Dagster Labs 在 2020 年开源，核心定位是解决数据工程中两个常见问题：开发体验割裂、生产运行不可观测。

核心概念是"数据资产"。你把数据集、机器学习模型、报表声明为 Python 函数，框架自动管理依赖执行。每条数据流的上下游关系由装饰器推导，不需要手动维护 DAG 配置文件。这和 Airflow 那种需要单独定义 DAG 的方式不同，资产天然自带血缘信息。

举个例子，定义三个关联的数据资产：拉取人口数据、训练变化模型、生成统计报表。

@dg.asset
def country_populations() -> pd.DataFrame:
    df = pd.read_html("https://tinyurl.com/mry64ebh")[0]
    df.columns = ["country", "pop2022", "pop2023", "change", "continent", "region"]
    df["change"] = df["change"].str.rstrip("%").astype("float")
    return df

@dg.asset
def continent_change_model(country_populations: pd.DataFrame) -> LinearRegression:
    data = country_populations.dropna(subset=["change"])
    return LinearRegression().fit(pd.get_dummies(data[["continent"]]), data["change"])

@dg.asset
def continent_stats(country_populations, continent_change_model) -> pd.DataFrame:
    result = country_populations.groupby("continent").sum()
    result["pop_change_factor"] = continent_change_model.coef_
    return result

函数的输入参数名对应上游资产，产出自动注册为下游可引用节点。在 Web UI 里这套依赖关系呈现为图谱，每个节点的运行时间、数据质量、血缘追溯都集中显示。

Dagster 在三个层面发挥作用。

第一层，开发平台。声明式定义资产，内置 CI/CD 实践。Asset checks 机制允许在数据产生点直接嵌入质量检验规则，开发阶段就能暴露缺陷。配合 dagster-dg-cli 工具，项目脚手架和代码生成都自动化，新建一个数据管线只需要几条命令。

第二层，编排引擎。多租户多工具的生产级调度，支持同时运行数千个任务。底层的计算引擎是 Spark、dbt 还是 Airbyte，Dagster 统一编排，每种工具的运行日志和指标集中在同一个视图。

第三层，控制平面。元数据、诊断、血缘集中管理。Assets 的上下游变更会自动检测，上游 schema 变动时下游立刻感知。性能和异常定位不需要跨系统查询。

生态集成方面，Dagster 覆盖了主流数据工具：Spark、dbt、Airbyte、Fivetran、Snowflake、Databricks 等。每个集成都包装为现成的库，pip 安装即可使用。

安装：

uv add dagster dagster-webserver dagster-dg-cli

项目采用 Apache 2.0 许可证，支持 Python 3.9 到 3.14。社区渠道包括 Slack、GitHub Discussions、技术博客、YouTube，官方提供完整教程和开发者文档，新手到贡献者都能找到对应资源。

Discussions、技术博客、YouTube，官方提供完整教程和开发者文档，新手到贡献者都能找到对应资源。