进阶 dagster.io 2026-10-10 01:48:56 · 6 阅读

第31章 Dagster 1.10 发布：并发控制、ELT 集成与用户体验全面升级

直觉式的并发控制、更完善的 ELT 集成以及开发者体验升级我们很高兴地宣布 Dagster 1.10 "Mambo No. 5" 正式发布！

此次发布聚焦于提升平台的清晰度和一致性，让现有功能更直观、易学且令人愉悦。我们显著优化了并发管理、ELT 集成和开发者体验等核心领域。

让我们深入看看本次版本的主要改进。

并发控制更新现代化的 ELT 集成开发者体验改进感谢自 1.9 版本以来所有社区贡献者的付出

直觉式的并发控制管理数据管道时最大的挑战之一是如何控制并发执行。在 Dagster 1.10 中，我们彻底重构了并发的运作方式，使其更易于理解、配置和监控。

并发改进

统一的并发池：不再为运行（run）和操作（operation）维护分离的 API，我们引入了“池”（pools）的概念，允许在不同粒度级别执行并发限制。增强的 UI 可见性：并发信息现在可直接在 UI 中轻松访问和配置。

# 改进前：使用标签作为并发组 @asset(tags={"dagster/concurrency_key": "heavy_process"}) def process_data(): pass

# 改进后：使用显式的并发参数 @asset(pool="heavy_process") def process_data(): pass

配置优化：并发设置现在在部署配置中逻辑地归入一个专用的并发块下。

这种新方法立即阐明了当前实施的并发控制措施及其对管道的影响。

现代化的 ELT 集成我们更新了与 Fivetran 和 Airbyte 等流行 ELT 工具的集成，以更好地利用 Dagster 的基于资产的模式，并提供增强的数据管道可见性。

关键改进

可定制的资产加载：新的翻译器（translator）模式允许对外部工具资产在 Dagster 中的表示进行细粒度控制。简化的接口：将多种配置方法整合为单一、直观的 get_asset_spec() 方法。

from dagster_fivetran import ( DagsterFivetranTranslator, FivetranConnectorTableProps, FivetranWorkspace, load_fivetran_asset_specs, )

import dagster as dg

fivetran_workspace = FivetranWorkspace( account_id=dg.EnvVar("FIVETRAN_ACCOUNT_ID"), api_key=dg.EnvVar("FIVETRAN_API_KEY"), api_secret=dg.EnvVar("FIVETRAN_API_SECRET"), )

class MyCustomFivetranTranslator(DagsterFivetranTranslator): def get_asset_spec(self, props: FivetranConnectorTableProps) -> dg.AssetSpec: default_spec = super().get_asset_spec(props) return default_spec.replace_attributes( key=default_spec.key.with_prefix("prefix"), ).merge_attributes(metadata={"custom": "metadata"})

fivetran_specs = load_fivetran_asset_specs( fivetran_workspace, dagster_fivetran_translator=MyCustomFivetranTranslator() )

defs = dg.Definitions(assets=fivetran_specs, resources={"fivetran": fivetran_workspace})

元数据可见性增强：更好地展示管道状态和性能数据。

开发者体验改进我们做了几项生活质量改进，使使用 Dagster 进行开发更加流畅：

gRPC 服务器整合 dagster dev 命令现在在每个代码位置启动单个集中式服务器，减少了计算开销并最小化进程相关的错误。

增强的资产 API

灵活的依赖声明：可以在 deps 参数和基于参数的依赖之间混合搭配。改进的 IO Manager 支持：现在与 multi_asset_check 无缝配合使用。按分区元数据：在物化期间为单个分区添加特定元数据。

@asset(deps=["upstream1", "upstream2"]) def my_asset(upstream2): # 仅需参数以访问 IO manager pass

@multi_asset_check( specs=[ AssetCheckSpec("check1", asset=asset1), AssetCheckSpec("check2", asset=asset2), ] ) def multi_check(asset1: int, asset2: int) -> Iterable[AssetCheckResult]: pass

新的模块加载 API 我们引入了新的 API 来简化从模块加载定义： import dagster as dg import package.submodule as my_submodule

defs = dg.load_definitions_from_module(my_submodule)

新的 API 生命周期我们引入了新的 API 生命周期阶段，配有独立的装饰器、注解和警告，以帮助开发者就项目中使用的 API 做出明智的决定。

预览（Preview）：处于开发和测试早期阶段的功能标记为预览。测试版（Beta）：仍在开发中，但已准备好在非关键生产环境中使用的功能标记为测试版。已取代（Superseded）：仍可用但不再被视为最佳实践的功能标记为已取代。

在我们的指南中查看所有 API 生命周期阶段。

展望未来 Dagster 1.10 体现了我们对让数据编排更直观、更强大的承诺。这些改进为今年稍后推出的令人兴奋的功能奠定了基础。

pip install -U dagster

想要尝试这些新功能吗？立即升级到 Dagster 1.10：

查看我们更新的文档以了解这些新功能和改进的更多详细信息。

一如既往，我们希望听到您的反馈！加入我们的 Slack 分享您对新版本的想法和体验。

致谢我们要感谢所有为促成此次发布而付出努力和创新的社区贡献者。

请留意未来版本中更多的更新和增强功能。有反馈或问题？在 Slack 或 Github 中发起讨论。有兴趣与我们共事吗？查看我们的空缺职位。喜欢这类内容吗？在 LinkedIn 上关注我们。