Limiting concurrency in data pipelines#

Concurrency is an essential concept in modern programming, particularly when working with data pipelines. While concurrency can improve the efficiency of your pipelines, too many processes executing at once can cause issues. Limiting concurrency in your data pipelines can help prevent performance problems and downtime.

By the end of this guide, you'll:

Understand the basics of concurrency
Learn about the options Dagster offers for limiting concurrency
Learn how to configure concurrency in Dagster
Understand how to troubleshoot run queueing issues

Understanding the basics#

Concurrency is the ability of a system to execute multiple processes in parallel. Before we go any further, let's go over the terminology in this guide:

Term	Definition
Op	The core unit of computation in Dagster, an op is a function that performs a task — for example, sending an email or kicking off a job in dbt.
Asset definition	An asset definition is a Dagster object that couples an asset to the function and upstream assets used to produce its contents. An asset is an object in persistent storage, such as a table, file, or persisted machine learning model.
Job	A job is a graph of ops or assets connected via data dependencies. Jobs are the main unit of execution and monitoring in Dagster.
Run	A run is a single execution of a job in Dagster.
Run queue	A sequence of Dagster runs waiting to be executed.
Executor	Executors are responsible for executing steps within a job run. Executors can range from single-process serial executors to managing per-step computational resources with a control plane.

Understanding Dagster’s concurrency limit options#

Dagster supports placing limits on the number of processes that can be in progress at a single time, at the run and op-level. Depending on your needs, one or both types may be suitable:

Run concurrency controls the total number of runs in a deployment that can execute at the same time. Run concurrency applies to both op and asset-based jobs and all code locations in a single deployment. Any runs beyond the limit will be queued and won’t use any compute.
Op/asset concurrency controls the number of ops/assets that can execute in parallel in a single run.
Global op/asset concurrency
(Experimental)
controls the number of ops/assets that can execute in parallel across all runs.

By limiting the number of parallel processes Dagster executes, you can ensure that resources are not overwhelmed and that each process has the resources it needs to run efficiently. This can lead to faster and more reliable pipeline execution, as well as easier monitoring and debugging of issues.

Configuring run-level concurrency#

Run concurrency can be configured by:

Limiting the overall number of runs in a deployment
Specifying limits using tags, which is useful for applying limits to specific jobs. For example, jobs that share a resource could all have the same tag.

These methods of limiting concurrency can be used individually or together. For each queued run, Dagster will check that launching the run will satisfy all conditions. For example, if launching a run would exceed the maximum number of concurrent runs, Dagster will queue the run until another run finishes.

Note: Run-level concurrency rules can't currently be used to prevent a run from being submitted to the queue; they only control limits for the number of runs in the queue that will be launched at once. If you want to prevent runs from ever being launched if another job is already running, check out this Github discussion for an example.

Limiting overall runs#

How you limit the overall number of concurrent runs in a deployment depends on whether you're using Dagster+ or Dagster Open Source:

Dagster+: Use the Dagster+ UI or the dagster-cloud CLI
Dagster Open Source: Use your instance's dagster.yaml

To enable this limit, use run_queue.max_concurrent_runs. For example, the following would limit the number of concurrent runs for the deployment to 15:


run_queue:
  max_concurrent_runs: 15

When defining a value for max_concurrent_runs, keep the following in mind:

This setting defaults to 10
Disable the setting with a -1 value, which means no limit will be enforced. Note: All other negative numbers are invalid, and disabling this setting isn't supported for Dagster+.
A value of 0 prevents any runs from launching

Limiting specific runs using tags#

How you limit specific runs based on tags depends on whether you're using Dagster+ or Dagster Open Source:

Dagster+: Use the Dagster+ UI or the dagster-cloud CLI
Dagster Open Source: Use your instance's dagster.yaml

To enable this limit, use run_queue.tag_concurrency_limits. This key accepts a list of tags and their corresponding concurrency limits.


run_queue:
  max_concurrent_runs: 15
  tag_concurrency_limits:
    - key: "database"
      value: "redshift" # applies when the `database` tag has a value of `redshift`
      limit: 4
    - key: "dagster/backfill" # applies when the `dagster/backfill` tag is present, regardless of value
      limit: 10

Let’s review what this configuration will do:

For runs with a database tag with a value of redshift, a maximum of four runs can execute concurrently
For runs with a dagster/backfill tag, a maximum of 10 runs can execute concurrently. Note that this implementation applies to any value of the dagster/backfill tag, whereas the database: redshift example only applies the limit when there’s a specific tag value.

Limiting runs by unique tag value#

To apply separate limits to each unique value of a tag, set a limit for each unique value using applyLimitPerUniqueValue:


run_queue:
  tag_concurrency_limits:
    - key: "use-case"
      value:
        applyLimitPerUniqueValue: true
      limit: 3

With this configuration, each unique value of the use-case tag will be limited to three concurrent runs. For example, team: marketing and team: sales will both be limited to three concurrent runs each.

Configuring op/asset-level concurrency#

Looking for global op/asset concurrency? Check out Dagster's experimental support for global op/asset concurrency limits. For more fine-grained control, you can also check out our integration with Celery.

Utilizing op/asset-level concurrency provides fine-grained control for the maximum number of ops/assets that can be executed at once within a single run, ensuring that shared resources aren't overwhelmed.

Op and asset-level concurrency is enabled by configuring a job executor. Using the executor, you can:

Limit the number of ops/assets that can execute in a single run at once
Specify limits using tags

Limiting overall concurrency in a job#

To limit concurrency for ops and assets in jobs, use max_concurrent in the job’s config, either in Python or using the Launchpad in the Dagster UI.

Note: The examples in this section use the multiprocess_executor, which uses the multiprocess key. This key will change depending on the type of executor you're using. Refer to the Job executor documentation for more info.

While max_concurrent is used to limit concurrency in both asset and op-based jobs, how jobs are defined differs between assets and ops. Click the tabs below for examples.

Asset-based jobs#

Asset jobs are defined using define_asset_job. In this example, using the multiprocess_executor, the job will execute up to three assets at once:


assets_job = define_asset_job(
    name="assets_job",
    config={
        "execution": {
            "config": {
                "multiprocess": {
                    "max_concurrent": 3,      # limits concurrent assets to 3
                },
            }
        }
    }
)

Op-based jobs#

Op-based jobs are defined using the @job decorator. In this example, using the multiprocess_executor, the job will execute up to four ops at once:


@job(
    config={
        "execution": {
            "config": {
                "multiprocess": {
                    "max_concurrent": 4,
                    "tag_concurrency_limits": [
                        {
                            "key": "database",
                            "value": "redshift",
                            "limit": 2,
                        }
                    ],
                },
            }
        }
    }
)
def tag_concurrency_job(): ...

Limiting concurrency using tags#

Limits can be specified for all ops/assets with a specific tag key or key-value pair. If any limit would be exceeded by launching an op/asset, then the op/asset will be queued.

To specify limits on the number of ops/assets with a specific tag, use tag_concurrency_limits in the job’s config, either in Python or using the Launchpad in the Dagster UI.

While tag_concurrency_limits is used to limit concurrency in both asset and op-based jobs, how jobs are defined differs between assets and ops. Click the tabs below for examples.

Asset-based jobs#

Unlike op-based jobs, asset jobs use the op_tags field on each asset when checking them for tag concurrency limits. In this example, using the multiprocess_executor, the job will execute up to three assets at once with the database tag equal to snowflake:


# example asset with tags, specified using op_tags
@asset(op_tags={"database": "snowflake"})
def asset1():
    ...

assets_job = define_asset_job(
    name="assets_job",
    config={
        "execution": {
            "config": {
                "multiprocess": {
                    "tag_concurrency_limits": [
                        {
                            "key": "database",
                            "value": "snowflake",
                            "limit": 3,
                        }
                    ],
                },
            }
        }
    }
)

Op-based jobs#

In this example, using the multiprocess_executor, the job will execute up to two ops at once with the database tag equal to redshift:


@job(
    config={
        "execution": {
            "config": {
                "multiprocess": {
                    "max_concurrent": 4,
                    "tag_concurrency_limits": [
                        {
                            "key": "database",
                            "value": "redshift",
                            "limit": 2,
                        }
                    ],
                },
            }
        }
    }
)
def tag_concurrency_job(): ...

Limiting op/asset concurrency across runs#

This feature is experimental and is only supported with Postgres/MySQL storages.

For specific ops/assets#

Limits can be specified on the Dagster instance using the special op tag dagster/concurrency_key. If this instance limit would be exceeded by launching an op/asset, then the op/asset will be queued.

For example, to globally limit the number of running ops touching Redshift to two, the op/asset must be first tagged with the global concurrency key:


@op(tags={"dagster/concurrency_key": "redshift"})
def my_redshift_op(): ...


@asset(op_tags={"dagster/concurrency_key": "redshift"})
def my_redshift_table(): ...

To specify limits on the number of ops/assets tagged with a particular concurrency key, you can either use the dagster CLI command or the Dagster UI:

To specify a global concurrency limit using the CLI, use:


dagster instance concurrency set <concurrency_key> <limit>

To specify a global concurrency limit using the Dagster UI, navigate to the Concurrency limits tab on the Deployment page.

The concurrency key should match the name that the op/asset is tagged with. For example, if the op/asset is tagged with dagster/concurrency_key: redshift, then the concurrency key should be redshift.

For all ops/assets#

A default concurrency limit can be configured for the instance, for any concurrency keys that do not have an explicit limit set:

Dagster+: Use the Dagster+ UI or the dagster-cloud CLI
Dagster Open Source: Use your instance's dagster.yaml

To enable this default value, use concurrency.default_op_concurrency_limit. For example, the following would set the default concurrency value for the deployment to 1:


concurrency:
  default_op_concurrency_limit: 1

Setting priority for ops/assets#

By default, slots will be allocated to op / asset execution on a first-in-first-out basis. However, steps from high priority runs will take precedence over steps from low priority runs, as determined by the priority tag on the individual run.


@job(tags={"dagster/priority": "3"})
def important_job(): ...


@schedule(
    cron_schedule="* * * * *",
    job_name="important_job",
    execution_timezone="US/Central",
    tags={"dagster/priority": "-1"},
)
def less_important_schedule(_): ...

Priority can also be assigned based on the priority tag of the op/asset definition.


@op(tags={"dagster/concurrency_key": "foo", "dagster/priority": "3"})
def my_op(): ...


@asset(op_tags={"dagster/concurrency_key": "foo", "dagster/priority": "3"})
def my_asset(): ...

When the priority tag is set on both the run and the asset/op definition, the step priority will be the sum of the two values.

Freeing concurrency slots#

With a global concurrency limit set, it is useful to configure your instance to automatically free slots for canceled/failed runs. Certain execution errors can cause steps to exit without freeing an occupied concurrency slot.

To automatically recover from this state, you can configure your instance's run monitoring to automatically free concurrency slots for runs that have reached some terminal state (e.g. failed or canceled runs).

In dagster.yaml, you can configure the interval between a run end and the slot clean up like so:


run_monitoring:
  enabled: true
  free_slots_after_run_end_seconds: 300 # free any hanging concurrency slots after 5 minutes from the end of a run

Throttling concurrency-limited runs#

If a global concurrency limit is set, some runs may be active without an executing op due to concurrency restraints. When this occurs, runs can unnecessarily utilize resources or make it difficult to see whether other runs are progressing.

You can throttle runs and block their step progression to mitigate this issue. To do this, configure the run queue to keep these runs in a QUEUED state until a slot becomes free for one of its steps:


run_queue:
  block_op_concurrency_limited_runs:
    enabled: true
    op_concurrency_slot_buffer: 1 # buffer of 1 means 1 run might be started even though there are no slots available

Troubleshooting#

When limiting concurrency, you might run into some issues until you get the configuration right.

Runs going to STARTED status and skipping QUEUED#

This only applies to Dagster Open Source.

The run_queue key may not be set in your instance's settings. In the Dagster UI, navigate to Deployment > Configuration and verify that the run_queue key is set.

Runs remaining in QUEUED status#

The possible causes for runs remaining in QUEUED status depend on whether you're using Dagster+ or Dagster Open Source.

If runs aren't being dequeued in Dagster+, the root causes could be:

If using a Hybrid deployment, the agent serving the deployment may be down. In this situation, runs will be paused.
Dagster+ is experiencing downtime. Check the status page for the latest on potential outages.

Limiting concurrency in data pipelines#

Understanding the basics#

Understanding Dagster’s concurrency limit options#

Configuring run-level concurrency#

Limiting overall runs#

Limiting specific runs using tags#

Limiting runs by unique tag value#

Configuring op/asset-level concurrency#

Limiting overall concurrency in a job#

Asset-based jobs#

Op-based jobs#

Limiting concurrency using tags#

Asset-based jobs#

Op-based jobs#

Limiting op/asset concurrency across runs#

For specific ops/assets#

For all ops/assets#

Setting priority for ops/assets#

Freeing concurrency slots#

Throttling concurrency-limited runs#

Troubleshooting#

Runs going to STARTED status and skipping QUEUED#

Runs remaining in QUEUED status#

Related#