进阶 dagster.io 2026-10-10 01:48:56 · 6 阅读
第34章 Dagster 1.13 发布：强化 AI 技能支持，新增分区资产检查与虚拟资产预览
Dagster 技能、分区资产检查、状态支持组件、虚拟资产以及更强的集成能力。Dagster 1.13 旨在让框架更易于采用：更快的原型开发、更简单的扩展，以及更便捷地从代码和 AI 工具中调用。
本次发布围绕以下几个主题展开：
- 提升对 AI 工具和编码代理的支持。
- 改进分区数据的建模和操作方式。
- 扩展组件库，并默认启用状态支持组件，使基于外部元数据的集成更易处理。
- 预览版虚拟资产支持，便于表示如数据库视图等逻辑资产。
以下是亮点内容：
AI 辅助 Dagster 开发支持升级
在 Dagster 1.13 中，我们简化了使用现代 AI 工具进行开发的过程。
我们发布了开源的 dagster-io/skills 仓库：这是一组专为 OpenAI Codex、Claude Code、OpenCode 及其他 LLM 框架设计的 Dagster 专用 AI 技能。不同于将 Dagster 指南绑定在某个特定的助手产品上，技能包将以可复用的格式公开 Dagster 专业知识，使其能跨越不同的智能编码工具使用。
这使得它们在整个 Dagster 项目生命周期中都非常有用：
- 原型开发新项目及定义
- 构建生产级系统，涵盖资产、自动化、结构和部署的最佳实践
- 故障排查、运行行为分析及配置问题诊断
你可以通过 npx skills 或 Claude 插件市场快速安装这些技能：
npx skills add dagster-io/skills
# 或在 Claude Code 中直接安装
/plugin marketplace add dagster-io/skills
/plugin install dagster-expert@dagster-skills
# 然后向你的编码代理询问类似的问题：
“帮我调试为什么物化失败了”
如果你已经在用 AI 工具构建和运维数据平台，这次发布将让 Dagster 更契合这一工作流。
分区资产检查上线
现已支持分区感知的资产检查，意味着检查可以评估上游资产的特定分区，而不是整个数据集。对于处理基于时间和其他类型分区资产的团队来说，这填补了一个重要空白：现在可以在与资产本身相同的粒度下表达数据质量和验证逻辑。
实际上，这意味着团队现在可以编写与分区数据实际生成和监控方式整齐对齐的检查：
import dagster as dg
daily_partitions = dg.DailyPartitionsDefinition(start_date="2025-01-01")
@dg.asset(partitions_def=daily_partitions)
def daily_orders(context: dg.AssetExecutionContext):
partition_key = context.partition_key
return load_orders_for_date(partition_key)
@dg.asset_check(
asset=daily_orders,
description="daily orders passed validation",
partitions_def=daily_partitions,
)
def daily_orders_check(daily_orders) -> dg.AssetCheckResult:
return dg.AssetCheckResult(
passed=len(daily_orders) > 0,
metadata={"row_count": len(daily_orders)},
)
本次发布还完善了分区工作流程的其余部分，使得分区选择及分区感知执行在日常使用中更加自然流畅。
通过 dg 提升开发者体验
dg CLI 及相关部署工作流获得了一系列生活质量改进。但更大的故事在于我们如何看待 AI 原生世界中的 CLI 工具。
我们日益意识到命令行工具是 AI 系统的强大抽象层：它们明确、可组合、可脚本化，且比临时的产品界面更容易被代理可靠调用。我们继续深化这一理念，通过扩展 dg api，使 Dagster 更多交互面可通过结构化命令访问。
这意味着代理（或人类！）可以通过专为程序化使用构建的稳定接口，更清晰地检查和操作资产、运行、计划、传感器、部署、秘密等 Dagster 平台的其他部分。
例如，代理可以通过如下简洁的 CLI 接口发现和检查作业：
$ dg api job list
NAME DESCRIPTION SCHEDULES SENSORS ASSET JOB
my_daily_job Daily snapshot 1 2 Yes
my_sensor_job Sensor-driven 0 1 No
$ dg api job get my_daily_job
Name: my_daily_job
Description: Daily snapshot
Asset Job: Yes
Tags: env=prod, team=data
Schedule: daily (0 0 * * *) [RUNNING]
Sensor: file_watcher [RUNNING]
对于人类用户，这项投入带来的是更清晰的工具一致的操作界面。对于使用 AI 助手的团队而言，这意味着 Dagster 在他们现有的工具中更容易检查、运维和故障排查。
默认启用状态支持组件
状态支持组件是 Dagster 用于那些定义依赖于外部元数据的集成的框架。无需在每次代码位置加载时重新查询外部系统，Dagster 可以在受控时间获取状态，持久化保存，然后从缓存结果构建定义。
对于数据团队，这意味着在使用 dbt、Fivetran、Airbyte、Tableau、Looker、Sigma、Power BI 等其他集成时，体验更具可预测性。
从 Dagster 1.13 开始，持久化本地状态成为状态支持组件的默认设置。对于这类组件，团队应将刷新定义状态作为 CI/CD 流程的一部分，以确保每次部署都基于最新的外部元数据，或按照定义的计划进行。如果使用 dg plus deploy configure 生成的 GitHub Actions 工作流，刷新步骤已默认包含在内。请参见关于在 CI/CD 中管理状态的文档。
虚拟资产预览支持
我们还添加了一项实验性功能：虚拟资产。
虽然 Dagster 允许对任意复杂的资产图进行建模，但历来难以准确表示和追踪那些无需显式计算即可融入父级变更的资产。
最典型的例子是数据库视图。只有在更改视图定义时才需要执行其功能，否则它会自动
它会在每次被查询时反映上游表的最新数据。但由于 Dagster 此前无法区分这类资产和普通资产，系统中的其他部分会出问题，比如 Declarative Automation 和过期状态计算，它们默认资产内容更新的唯一方式就是发生一次物化（materialization）事件。 现在，定义资产时可以传入 is_virtual=True，表示该资产的内容会在其父资产变化时自动更新： @dg.asset(is_virtual=True, deps=[upstream_table]) def my_view(context: dg.AssetExecutionContext, db: MyDatabase): db.execute("CREATE OR REPLACE VIEW my_view AS SELECT * FROM upstream_table") 如果你希望自动化条件能感知依赖链中的虚拟资产，可以用 resolve_through_virtual() 穿透它们，追溯到非虚拟的祖先资产： dg.AutomationCondition.eager().resolve_through_virtual() 我们还以可选方式支持自动把 dbt 项目中定义的所有视图（view）当作虚拟资产处理： type: dagster_dbt.DbtProjectComponent attributes: translator: enable_dbt_views_as_virtual_assets: true 该功能目前处于预览阶段，在我们收集反馈并可能对 API 和功能进行调整期间，不建议在生产环境中使用。期待看到你的用法！ 借助 20 多个新组件让上手更轻松 Dagster 1.13 继续推进声明式、易集成的流水线构建方式。在 1.12.x 系列中，我们新增了 20 多个组件，让 Dagster 更容易接入团队已经在用的工具、数仓、云服务和运维系统。 在组件方面，本周期新增或扩展了对以下内容的支持： DbtCloudComponent，用于把 dbt Cloud 项目加载为 Dagster 资产。 Spark 的声明式流水线支持（功能预览）。 针对 Blob Storage 和 ADLS2 的 Azure 资源组件。 针对 BigQuery、GCS、GCS File Manager 和 Dataproc 的 GCP 资源组件。 Databricks、Tableau、Looker、Census、Polytomic 等工具新增了组件和资产加载能力。 多个集成还获得了更深入的运维支持： dbt_cloud_assets 现在支持分区资产。 DatabricksAssetBundleComponent 更加灵活，包括作业级别的子集筛选和更好的资产映射。 DatabricksWorkspaceComponent 可以在 Dagster 运行被终止时自动取消对应的 Databricks 作业。 Fivetran 新增了用于可观测性的轮询 sensor、更丰富的连接器元数据、重新调度时自动重试，以及重新同步支持。 BI 集成现在会自动为资产补充更多表和存储相关的元数据。 这些改动呈现出一个清晰的趋势：集成不只是数量在增加，还在变得更可观测、更可配置，也更容易融入统一的 Dagster 模型。这既降低了上手门槛，也给团队留出了向更复杂部署演进的空间。 一个特别实用的例子是基于表元数据的自动匹配改进。随着越来越多的集成自动输出 dagster/table_name 元数据，Dagster 能更好地识别指向同一张底层表的资产，即使它们的 asset key 不同。对于需要在 ingestion 工具、BI 工具、数仓和代码定义的转换之间整合资产的团队来说，这减少了手动配置，让跨系统的血缘关系更加自然。 from dagster import TableMetadataSet @dg.asset( key=dg.AssetKey(["raw", "customers"]), metadata={TableMetadataSet(table_name="analytics.raw.customers")}, ) def raw_customers(): ... @dg.asset( deps=[ dg.AssetDep( "upstream_table", metadata={TableMetadataSet(table_name="analytics.raw.customers")}, ) ] ) def customer_health_report(): ... 在这个例子里，Dagster 能推断出 customer_health_report 依赖于 raw_customers，因为它们指向同一张表，尽管二者的 asset key 并不一致。 Dagster+ 的改进 本周期 Dagster+ 也有不少实质性升级，包括组织级的时区设置、面向 Pro 账户的服务用户（service users）、更健壮的代码服务器重新部署行为、部署和故障恢复期间 agent 行为的改进，以及 insights 和告警工作流的持续扩展。 对于在生产环境使用 Dagster+ 的团队，这些改动既提升了运维体验，也让作业、代码位置和告警的可观测性质量更上一层楼。 展望未来 有了 Dagster 1.13，数据团队应该能更快地采用 Dagster，并在采用之后跑得更快。从 Dagster Skills 和 AI 辅助工作流，到分区资产检查，再到 20 多个新组件，这次发布的目标就是让 Dagster 更好地服务于数据工程师和分析师每天的实际工作。 升级到最新的 1.13 版本试试看，并告诉我们你用它构建了什么！🚀 有反馈或问题？欢迎在 Slack 或 GitHub 上发起讨论。有兴趣和我们一起工作？看看我们的在招职位。想看更多这类内容？请在 LinkedIn 上关注我们。
它会在每次被查询时反映上游表的最新数据。但由于 Dagster 此前无法区分这类资产和普通资产，系统中的其他部分会出问题，比如 Declarative Automation 和过期状态计算，它们默认资产内容更新的唯一方式就是发生一次物化（materialization）事件。 现在，定义资产时可以传入 is_virtual=True，表示该资产的内容会在其父资产变化时自动更新： @dg.asset(is_virtual=True, deps=[upstream_table]) def my_view(context: dg.AssetExecutionContext, db: MyDatabase): db.execute("CREATE OR REPLACE VIEW my_view AS SELECT * FROM upstream_table") 如果你希望自动化条件能感知依赖链中的虚拟资产，可以用 resolve_through_virtual() 穿透它们，追溯到非虚拟的祖先资产： dg.AutomationCondition.eager().resolve_through_virtual() 我们还以可选方式支持自动把 dbt 项目中定义的所有视图（view）当作虚拟资产处理： type: dagster_dbt.DbtProjectComponent attributes: translator: enable_dbt_views_as_virtual_assets: true 该功能目前处于预览阶段，在我们收集反馈并可能对 API 和功能进行调整期间，不建议在生产环境中使用。期待看到你的用法！ 借助 20 多个新组件让上手更轻松 Dagster 1.13 继续推进声明式、易集成的流水线构建方式。在 1.12.x 系列中，我们新增了 20 多个组件，让 Dagster 更容易接入团队已经在用的工具、数仓、云服务和运维系统。 在组件方面，本周期新增或扩展了对以下内容的支持： DbtCloudComponent，用于把 dbt Cloud 项目加载为 Dagster 资产。 Spark 的声明式流水线支持（功能预览）。 针对 Blob Storage 和 ADLS2 的 Azure 资源组件。 针对 BigQuery、GCS、GCS File Manager 和 Dataproc 的 GCP 资源组件。 Databricks、Tableau、Looker、Census、Polytomic 等工具新增了组件和资产加载能力。 多个集成还获得了更深入的运维支持： dbt_cloud_assets 现在支持分区资产。 DatabricksAssetBundleComponent 更加灵活，包括作业级别的子集筛选和更好的资产映射。 DatabricksWorkspaceComponent 可以在 Dagster 运行被终止时自动取消对应的 Databricks 作业。 Fivetran 新增了用于可观测性的轮询 sensor、更丰富的连接器元数据、重新调度时自动重试，以及重新同步支持。 BI 集成现在会自动为资产补充更多表和存储相关的元数据。 这些改动呈现出一个清晰的趋势：集成不只是数量在增加，还在变得更可观测、更可配置，也更容易融入统一的 Dagster 模型。这既降低了上手门槛，也给团队留出了向更复杂部署演进的空间。 一个特别实用的例子是基于表元数据的自动匹配改进。随着越来越多的集成自动输出 dagster/table_name 元数据，Dagster 能更好地识别指向同一张底层表的资产，即使它们的 asset key 不同。对于需要在 ingestion 工具、BI 工具、数仓和代码定义的转换之间整合资产的团队来说，这减少了手动配置，让跨系统的血缘关系更加自然。 from dagster import TableMetadataSet @dg.asset( key=dg.AssetKey(["raw", "customers"]), metadata={TableMetadataSet(table_name="analytics.raw.customers")}, ) def raw_customers(): ... @dg.asset( deps=[ dg.AssetDep( "upstream_table", metadata={TableMetadataSet(table_name="analytics.raw.customers")}, ) ] ) def customer_health_report(): ... 在这个例子里，Dagster 能推断出 customer_health_report 依赖于 raw_customers，因为它们指向同一张表，尽管二者的 asset key 并不一致。 Dagster+ 的改进 本周期 Dagster+ 也有不少实质性升级，包括组织级的时区设置、面向 Pro 账户的服务用户（service users）、更健壮的代码服务器重新部署行为、部署和故障恢复期间 agent 行为的改进，以及 insights 和告警工作流的持续扩展。 对于在生产环境使用 Dagster+ 的团队，这些改动既提升了运维体验，也让作业、代码位置和告警的可观测性质量更上一层楼。 展望未来 有了 Dagster 1.13，数据团队应该能更快地采用 Dagster，并在采用之后跑得更快。从 Dagster Skills 和 AI 辅助工作流，到分区资产检查，再到 20 多个新组件，这次发布的目标就是让 Dagster 更好地服务于数据工程师和分析师每天的实际工作。 升级到最新的 1.13 版本试试看，并告诉我们你用它构建了什么！🚀 有反馈或问题？欢迎在 Slack 或 GitHub 上发起讨论。有兴趣和我们一起工作？看看我们的在招职位。想看更多这类内容？请在 LinkedIn 上关注我们。
