Dagster 1.10: Concurrency + UX Upgrades

Intuitive Concurrency Controls, Improved ELT integrations, and Developer Experience Upgrades

We're excited to announce the release of Dagster 1.10 Mambo No. 5!

This release focuses on bringing clarity and consistency across the platform, making existing features more intuitive and enjoyable to learn. We've significantly improved core areas like concurrency management, ELT integrations, and developer experience.

Let's dive into the major improvements in this release.

Concurrency Updates
Modernized ELT Integrations
Developer Experience Improvements
And thanks to all the community contributors since 1.9

Intuitive Concurrency Controls

One of the biggest challenges in managing data pipelines is controlling concurrent execution. With Dagster 1.10, we've completely revamped how concurrency works to make it easier to understand, configure, and monitor.

Concurrency Improvements

Unified Concurrency Pools: Rather than having separate APIs for run and operation concurrency, we've introduced the concept of concurrency "pools" that can be enforced at different granularities.
Enhanced UI Visibility: Concurrency information is now easily accessible and configurable directly in the UI.

# Before: Using tags for concurrency groups
@asset(tags={"dagster/concurrency_key": "heavy_process"})
def process_data():
    pass

# After: Using explicit concurrency argument
@asset(pool="heavy_process")
def process_data():
    pass

Improved Configuration: Concurrency settings are now logically grouped under a dedicated concurrency block in deployment settings.

The new approach immediately clarifies what concurrency controls are in place and how they affect your pipelines.

Modernized ELT Integrations

We've refreshed our integrations with popular ELT tools like Fivetran and Airbyte to leverage Dagster's asset-based patterns better and provide enhanced visibility into your data pipelines.

Key Improvements

Customizable Asset Loading: A new translator pattern allows fine-grained control over how external tool assets are represented in Dagster
Simplified Interfaces: Consolidated multiple configuration methods into a single, intuitive get_asset_spec() method

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})

Enhanced Metadata Visibility: Better surfacing of pipeline status and performance data

Developer Experience Improvements

We've made several quality-of-life improvements to make development with Dagster more seamless:

GRPC Server Consolidation

The dagster dev command now launches a single centralized server per code location, reducing computational overhead and minimizing process-related errors.

Enhanced Asset APIs

Flexible Dependency Declaration: Mix and match between deps parameter and argument-based dependencies
Improved IO Manager Support: Now works seamlessly with multi_asset_check
Per-Partition Metadata: Add specific metadata to individual partitions during materialization

@asset(deps=["upstream1", "upstream2"])
def my_asset(upstream2):  # Only need argument for IO manager access
    pass

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

New Module Loading APIs

We've introduced new APIs to simplify loading definitions from modules:

import dagster as dg
import package.submodule as my_submodule

defs = dg.load_definitions_from_module(my_submodule)

New API lifecycle

We've introduced new API lifecycle stages, with their own decorators, annotations and warnings, to help developers make informed decisions about which APIs to use in their projects.

Preview: Features at an early stage of development and testing are marked as preview.
Beta: Features that are still under development, but are ready for use in non-critical production environments, are marked as beta.
Superseded: Features that are still available, but no longer considered best practice, are marked as superseded.

Discover all API lifecycle stages in our guide.

Looking Forward

Dagster 1.10 represents our commitment to making data orchestration more intuitive and powerful. These improvements lay the groundwork for exciting features coming later this year.