Asset Patterns Reference

Asset Patterns Reference

Pattern Summary

Launching Assets

Once you've defined assets, use the dg launch command to materialize them. For comprehensive documentation on launching assets, see the CLI launch reference.

Basic Asset Definition

import dagster as dg

@dg.asset
def my_asset() -> None:
    """
    Docstring becomes the asset description in the UI.
    """
    # Your computation logic
    pass

Key points:

• Function name becomes the asset key
• Docstring becomes the description
• Return type annotation is optional but recommended

Asset Dependencies

Parameter-Based Dependencies

When an asset depends on another Dagster-managed asset, use parameters:

@dg.asset
def upstream_asset() -> dict:
    return {"data": [1, 2, 3]}

@dg.asset
def downstream_asset(upstream_asset: dict) -> list:
    """
    Receives the return value of upstream_asset automatically.
    """
    return upstream_asset["data"]

How it works:

• Parameter name must match the upstream asset's function name
• Dagster automatically passes the materialized output
• Creates a visual dependency in the asset graph

External Dependencies with deps=

When an asset depends on something not managed by Dagster or without a return value:

@dg.asset(deps=["external_table", "raw_file"])
def processed_data() -> None:
    """
    Depends on external_table and raw_file but doesn't receive their values.
    """
    # Read from external sources directly
    pass

Use deps= when:

• The upstream asset doesn't return a value (returns None)
• The asset is external (file created by another process)
• You need loose coupling between assets

Mixed Dependencies

Combine both patterns when needed:

@dg.asset(deps=["raw_file"])
def enriched_data(reference_table: dict) -> dict:
    """
    Depends on:
    - raw_file (via deps=, doesn't receive value)
    - reference_table (via parameter, receives value)
    """
    # Read raw_file from disk, enrich with reference_table
    return {"enriched": reference_table}

Asset Metadata

Definition Metadata (Static)

Applied once when the asset is defined:

@dg.asset(
    description="Detailed description for the UI",
    group_name="analytics",
    key_prefix=["warehouse", "staging"],
    owners=["team:data-engineering", "user@example.com"],
    tags={"priority": "high", "pii": "true", "domain": "sales"},
    code_version="1.2.0",
)
def my_asset() -> None:
    pass

Best Practices for Metadata:

• owners: Specify team (team:name) or individuals for accountability
• tags: Primary organizational mechanism—use liberally for filtering and grouping
• code_version: Track when asset logic changes for lineage and debugging
• description: Explain what the asset represents and its business purpose
• group_name: Visual organization in UI (use for layers or domains)
• key_prefix: Hierarchical namespacing for multi-tenant or layered architectures

Materialization Metadata (Dynamic)

Captured each time the asset materializes:

import dagster as dg

@dg.asset
def my_asset() -> dg.MaterializeResult:
    """Asset that reports metadata on each run."""
    data = fetch_data()
    row_count = len(data)

    # Save data...

    return dg.MaterializeResult(
        metadata={
            "row_count": dg.MetadataValue.int(row_count),
            "last_updated": dg.MetadataValue.text(str(datetime.now())),
            "sample_data": dg.MetadataValue.json(data[:5]),
        }
    )

MetadataValue Types

Asset Groups

Organize related assets visually in the UI:

@dg.asset(group_name="raw_data")
def raw_orders() -> None:
    pass

@dg.asset(group_name="raw_data")
def raw_customers() -> None:
    pass

@dg.asset(group_name="analytics")
def daily_revenue(raw_orders) -> None:
    pass

Best practices:

• Group by data layer: raw, staging, analytics, mart
• Group by domain: sales, marketing, finance
• Group by source: postgres, api, files

Asset Key Prefixes

Namespace assets for organization:

@dg.asset(key_prefix=["warehouse", "raw"])
def orders() -> None:
    """Asset key becomes: warehouse/raw/orders"""
    pass

@dg.asset(key_prefix=["warehouse", "staging"])
def orders_cleaned() -> None:
    """Asset key becomes: warehouse/staging/orders_cleaned"""
    pass

Use prefixes for:

• Multi-tenant architectures
• Environment separation
• Data layer organization (bronze/silver/gold)

Asset with Execution Context

Access runtime information during materialization:

@dg.asset
def context_aware_asset(context: dg.AssetExecutionContext) -> None:
    context.log.info("Starting asset materialization")

    # Access asset key
    asset_key = context.asset_key

    # Access partition key (if partitioned)
    if context.has_partition_key:
        partition = context.partition_key
        context.log.info(f"Processing partition: {partition}")

    # Access run ID
    run_id = context.run_id

Asset with Configuration

Make assets configurable at runtime:

from dagster import Config

class MyAssetConfig(Config):
    limit: int = 100
    include_archived: bool = False
    source_path: str

@dg.asset
def configurable_asset(config: MyAssetConfig) -> None:
    """
    Run with specific configuration values.
    """
    data = load_data(
        path=config.source_path,
        limit=config.limit,
        include_archived=config.include_archived,
    )

Asset Return Types

Returning Data Directly

@dg.asset
def returns_data() -> dict:
    return {"key": "value"}

Returning MaterializeResult

@dg.asset
def returns_result() -> dg.MaterializeResult:
    # Do work...
    return dg.MaterializeResult(
        metadata={"rows": 100}
    )

Returning Output with Type

from dagster import Output

@dg.asset
def returns_output() -> Output[dict]:
    data = {"key": "value"}
    return Output(
        value=data,
        metadata={"size": len(data)},
    )

Multi-Asset Pattern

Define multiple assets from one function:

from dagster import multi_asset, AssetOut

@multi_asset(
    outs={
        "users": AssetOut(),
        "orders": AssetOut(),
    }
)
def load_data():
    users_df = fetch_users()
    orders_df = fetch_orders()

    yield Output(users_df, output_name="users")
    yield Output(orders_df, output_name="orders")

Use when:

• One computation produces multiple logical assets
• Assets are always created together
• Shared setup is expensive

Graph Assets

@graph_asset Pattern

Use when multiple steps are needed to produce a single asset:

from dagster import graph_asset, op

@op
def fetch_data() -> dict:
    return {"raw": [1, 2, 3]}

@op
def transform_data(data: dict) -> dict:
    return {"processed": [x * 2 for x in data["raw"]]}

@op
def validate_data(data: dict) -> dict:
    assert len(data["processed"]) > 0
    return data

@graph_asset
def complex_asset():
    """Encapsulates multi-step logic into a single asset."""
    raw = fetch_data()
    processed = transform_data(raw)
    return validate_data(processed)

Use When:

• Single asset requires multiple distinct steps
• You want to encapsulate complexity
• Testing individual steps is important
• Steps are reusable across multiple assets

@graph_multi_asset Pattern

Use for complex pipelines producing multiple assets:

from dagster import graph_multi_asset, AssetOut

@graph_multi_asset(
    outs={
        "users": AssetOut(),
        "orders": AssetOut(),
    }
)
def etl_pipeline():
    """Multi-step pipeline producing multiple assets."""
    raw_data = extract_from_api()
    cleaned = clean_data(raw_data)
    users_out = extract_users(cleaned)
    orders_out = extract_orders(cleaned)

    return {"users": users_out, "orders": orders_out}

Use When:

• Multiple assets require shared complex logic
• Steps are expensive and should be shared
• Better encapsulation than separate assets with deps

Asset Factories

Generate similar assets programmatically:

def create_table_asset(table_name: str, schema: str):
    @dg.asset(
        name=f"{schema}_{table_name}",
        group_name=schema,
    )
    def _asset() -> None:
        load_table(schema, table_name)

    return _asset

# Generate assets
customers = create_table_asset("customers", "sales")
products = create_table_asset("products", "catalog")
orders = create_table_asset("orders", "sales")

Asset Selection Syntax

Select subsets of assets for materialization or job definition:

Basic Selection

from dagster import AssetSelection

# Select specific assets
AssetSelection.assets("asset_a", "asset_b", "asset_c")

# Select all assets
AssetSelection.all()

# Select by group
AssetSelection.groups("analytics", "raw_data")

# Select by key prefix
AssetSelection.key_prefixes(["warehouse", "staging"])

# Select by tag
AssetSelection.tag("priority", "high")

Dependency-Based Selection

# Select asset and all upstream dependencies
AssetSelection.assets("final_report").upstream()

# Select asset and all downstream dependencies
AssetSelection.assets("raw_data").downstream()

# Select asset and immediate upstream only
AssetSelection.assets("final_report").upstream(depth=1)

Combining Selections

# Union: assets in A OR B
selection_a | selection_b

# Intersection: assets in A AND B
selection_a & selection_b

# Difference: assets in A but not in B
selection_a - selection_b

# Example: All analytics assets except one
AssetSelection.groups("analytics") - AssetSelection.assets("excluded_asset")

Using in Jobs

import dagster as dg

analytics_job = dg.define_asset_job(
    name="analytics_job",
    selection=AssetSelection.groups("analytics").downstream(),
)

Using in CLI

# Materialize specific assets
dg launch --assets asset_a,asset_b

# Materialize all assets
dg launch --assets "*"

# Materialize by group (requires selection in job)
dg launch --job analytics_job

Common Anti-Patterns

References

• Assets API
• Asset Metadata
• Multi-Assets