Dagster 并发池：限制共享资源的并发访问-未完纪

为并发池设置默认限制

Dagster+：通过 Dagster+ 界面或 dagster-cloud 命令行工具，在部署设置中修改 concurrency 配置。

Dagster Open Source：使用实例的 dagster.yaml：

concurrency:
  pools:
    default_limit: 1

限制所有运行中正在执行的资产或 op 数量

可以把资产和 op 分配给并发池，以限制所有运行中正在执行的 op 数量。这很适合保护数据库、API 等共享资源。先用 pool 关键字参数将资产或 op 分配到池。

src/<project_name>/defs/assets.py：

import time

import dagster as dg


@dg.asset(pool="database")
def query_customers(context: dg.AssetExecutionContext):
    """Asset assigned to the 'database' pool."""
    context.log.info("Querying customers table...")
    time.sleep(5)  # Simulate database query
    return {"count": 1000}


@dg.asset(pool="database")
def query_orders(context: dg.AssetExecutionContext):
    """Asset assigned to the 'database' pool."""
    context.log.info("Querying orders table...")
    time.sleep(5)  # Simulate database query
    return {"count": 5000}


@dg.asset(pool="api")
def fetch_external_data(context: dg.AssetExecutionContext):
    """Asset assigned to the 'api' pool."""
    context.log.info("Fetching from external API...")
    time.sleep(3)  # Simulate API call
    return {"status": "success"}

在 Dagster 界面中查看资产或 op 的详情面板，可以核对池是否设置正确。

分配到池之后，可以通过界面或 dagster 命令行工具为部署配置该池的限制。

通过界面给 database 池指定限制：进入 Deployments → Concurrency 设置页，点击 Add pool limit。

通过命令行指定限制：

dagster instance concurrency set database 1

限制包含一组 op 的运行数量

并发池也能限制包含这些资产或 op、正在进行的运行数量。先按上一节把资产与 op 分配到池，并设置限制。

随后修改部署设置，调整池的执行粒度。若要限制任意时刻包含某个 op 的运行总数，而不是正在执行的 op 总数，需要将池粒度设为 run。

Dagster Core 将下面的内容加入 dagster.yaml；Dagster+ 将它加入部署设置：

concurrency:
  pools:
    granularity: 'run'

没有设置粒度时，默认粒度为 op。例如 foo 池限制为1，会确保所有运行中任意时刻只有一个 op 正在执行，但不会限制正在进行的运行数量。

清理被取消运行占用的并发槽位

默认情况下，运行取消或失败时，Dagster 不会自动释放并发池槽位。如果运行在持有槽位期间被取消，该槽位会一直被占用，阻止后续运行申请。若池限制为1，一个被取消的运行就可能使该池的所有后续运行永久阻塞。

为避免这个问题，应在运行监控设置中配置 free_slots_after_run_end_seconds，启用自动清理：

run_monitoring:
  enabled: true
  free_slots_after_run_end_seconds: 300

使用并发池时，原文强烈推荐配置这个选项。如果已经发生阻塞，参见并发问题排查指南中的恢复步骤。

来源与许可

作者：Dagster Labs 与贡献者。原文：Concurrency pools。原文入口：https://dagster.io/docs/guides/operate/managing-concurrency/concurrency-pools；上述docs.dagster.io页面为官方文档路径。Copyright 2026 Dagster Labs。本文为中文翻译，格式整理不改变API与命令。示例代码模拟数据库和 API 调用。

Dagster 开源仓库采用 Apache License 2.0。本稿附带该许可的来源链接，发布交接须保留完整许可和适用版权通知。配图为官方原文截图。

本文适用于 Dagster 1.13.25。

文章版权归作者所有，未经允许请勿转载。

THE END