Dagster Assets

Dagster Software-Defined Assets

When to Use

Activate this skill when the task involves:

• Defining data pipelines as Software-Defined Assets in Dagster
• Setting up partitioned assets (time-based, static, dynamic)
• Writing sensors to trigger asset materializations on events
• Implementing IO managers for custom storage backends
• Configuring declarative automation (AutomationCondition / eager / on_cron)
• Structuring Dagster projects, resources, and definitions
• Deploying Dagster with Docker, Kubernetes, or Dagster Cloud

Core Model

Asset (data object in storage)
  ↑
  │  materialized by
  │
Asset Function  ──reads──►  Upstream Assets (via deps / function params)
  │
  └─ uses IO Manager to write output
       uses Resources for external connections
       may be Partitioned (date/region/etc.)
       may be triggered by Sensor or Schedule or AutomationCondition

Every asset represents a persistent data object (table, file, ML model). The job of the asset function is to compute and save it. Dagster tracks materialization history, staleness, and lineage automatically.

Installation

pip install dagster dagster-webserver
# Optional integrations
pip install dagster-aws dagster-dbt dagster-spark dagster-k8s

Basic Assets

@asset — Single Asset

import dagster as dg
import pandas as pd

@dg.asset
def raw_orders() -> pd.DataFrame:
    """Load raw orders from source system."""
    return pd.read_csv("s3://data-lake/raw/orders.csv")

@dg.asset(deps=[raw_orders])
def stg_orders(raw_orders: pd.DataFrame) -> pd.DataFrame:
    """Clean and validate orders."""
    df = raw_orders.dropna(subset=["order_id", "customer_id"])
    df["order_date"] = pd.to_datetime(df["order_date"])
    df = df[df["total"] > 0]
    return df

@dg.asset
def fct_orders(stg_orders: pd.DataFrame) -> pd.DataFrame:
    """Build orders fact table."""
    return stg_orders.assign(
        revenue_usd=stg_orders["total"] / 100,
        order_month=stg_orders["order_date"].dt.to_period("M").astype(str),
    )

Asset with Context and Metadata

@dg.asset(
    group_name="silver",
    key_prefix=["warehouse", "silver"],
    compute_kind="pandas",
    tags={"team": "data-platform"},
)
def customers(context: dg.AssetExecutionContext, raw_customers: pd.DataFrame) -> pd.DataFrame:
    context.log.info(f"Processing {len(raw_customers)} raw customer rows")

    df = raw_customers.copy()
    df["email"] = df["email"].str.lower().str.strip()
    df = df.drop_duplicates(subset=["customer_id"])

    context.add_output_metadata({
        "row_count": len(df),
        "preview": dg.MetadataValue.md(df.head().to_markdown()),
        "schema": dg.MetadataValue.json(dict(df.dtypes.astype(str))),
    })
    return df

@multi_asset — Multiple Outputs from One Operation

@dg.multi_asset(
    outs={
        "users":  dg.AssetOut(key_prefix="silver", description="Cleaned users"),
        "events": dg.AssetOut(key_prefix="silver", description="Cleaned events"),
    },
    can_subset=True,
)
def parse_api_response(context: dg.AssetExecutionContext):
    """Single API call producing two tables."""
    data = fetch_from_api()

    users_df = pd.DataFrame(data["users"])
    events_df = pd.DataFrame(data["events"])

    yield dg.MaterializeResult(
        asset_key="users",
        metadata={"row_count": len(users_df)},
        value=users_df,
    )
    yield dg.MaterializeResult(
        asset_key="events",
        metadata={"row_count": len(events_df)},
        value=events_df,
    )

External Asset (Not Managed by Dagster)

raw_salesforce = dg.AssetSpec(
    key="raw_salesforce_accounts",
    group_name="external",
    description="Salesforce accounts loaded by Fivetran — not materialized by Dagster",
)

Partitioned Assets

Daily Partitions

daily_partitions = dg.DailyPartitionsDefinition(
    start_date="2024-01-01",
    timezone="UTC",
)

@dg.asset(partitions_def=daily_partitions)
def daily_events(context: dg.AssetExecutionContext) -> pd.DataFrame:
    partition_date = context.partition_key           # "2024-03-15"
    start, end = context.partition_time_window       # (datetime, datetime)

    df = query_warehouse(
        f"SELECT * FROM events WHERE event_date = '{partition_date}'"
    )
    context.log.info(f"Loaded {len(df)} events for {partition_date}")
    return df

Other Time Granularities

hourly    = dg.HourlyPartitionsDefinition(start_date="2024-01-01-00:00")
weekly    = dg.WeeklyPartitionsDefinition(start_date="2024-01-01")
monthly   = dg.MonthlyPartitionsDefinition(start_date="2024-01-01")

Static Partitions

region_partitions = dg.StaticPartitionsDefinition(["us", "eu", "apac"])

@dg.asset(partitions_def=region_partitions)
def regional_summary(context: dg.AssetExecutionContext) -> pd.DataFrame:
    region = context.partition_key      # "us" | "eu" | "apac"
    return compute_summary(region=region)

Multi-Dimensional Partitions

multi_partitions = dg.MultiPartitionsDefinition({
    "date":   dg.DailyPartitionsDefinition(start_date="2024-01-01"),
    "region": dg.StaticPartitionsDefinition(["us", "eu", "apac"]),
})

@dg.asset(partitions_def=multi_partitions)
def regional_daily(context: dg.AssetExecutionContext) -> pd.DataFrame:
    keys = context.partition_key.keys_by_dimension   # {"date": "2024-03-15", "region": "us"}
    return compute(date=keys["date"], region=keys["region"])

Dynamic Partitions (Runtime)

customer_partitions = dg.DynamicPartitionsDefinition(name="customers")

@dg.asset(partitions_def=customer_partitions)
def customer_report(context: dg.AssetExecutionContext) -> pd.DataFrame:
    customer_id = context.partition_key
    return build_report(customer_id=customer_id)

Add partitions programmatically (usually from a sensor):

context.instance.add_dynamic_partitions("customers", ["cust_001", "cust_002"])

Resources

Resources encapsulate external connections. Inject them into asset functions.

from dagster_aws.s3 import S3Resource
import dagster as dg

class WarehouseResource(dg.ConfigurableResource):
    host:     str
    database: str
    user:     str
    password: str

    def query(self, sql: str) -> pd.DataFrame:
        import psycopg2
        conn = psycopg2.connect(
            host=self.host, dbname=self.database,
            user=self.user, password=self.password,
        )
        return pd.read_sql(sql, conn)

@dg.asset
def orders(warehouse: WarehouseResource) -> pd.DataFrame:
    return warehouse.query("SELECT * FROM raw.orders")

# Wire up in Definitions
defs = dg.Definitions(
    assets=[orders],
    resources={
        "warehouse": WarehouseResource(
            host="db.prod.internal",
            database="analytics",
            user=dg.EnvVar("DB_USER"),
            password=dg.EnvVar("DB_PASSWORD"),
        ),
        "s3": S3Resource(region_name="us-east-1"),
    },
)

IO Managers

IO managers control how asset outputs are stored and loaded. Default manager: in-memory (no persistence) or filesystem pickle.

Custom IO Manager — Parquet on S3

from dagster import IOManager, InputContext, OutputContext, io_manager
import boto3
import pandas as pd
import io

class S3ParquetIOManager(IOManager):
    def __init__(self, bucket: str, prefix: str = "data"):
        self.bucket = bucket
        self.prefix = prefix
        self.s3 = boto3.client("s3")

    def _key(self, context) -> str:
        parts = list(context.asset_key.path)
        if context.has_partition_key:
            parts.append(context.partition_key)
        return f"{self.prefix}/{'/'.join(parts)}.parquet"

    def handle_output(self, context: OutputContext, obj: pd.DataFrame):
        key = self._key(context)
        buf = io.BytesIO()
        obj.to_parquet(buf, index=False)
        buf.seek(0)
        self.s3.put_object(Bucket=self.bucket, Key=key, Body=buf.read())
        context.add_output_metadata({"s3_key": key, "rows": len(obj)})

    def load_input(self, context: InputContext) -> pd.DataFrame:
        key = self._key(context.upstream_output)
        response = self.s3.get_object(Bucket=self.bucket, Key=key)
        return pd.read_parquet(io.BytesIO(response["Body"].read()))

@io_manager(config_schema={"bucket": str, "prefix": str})
def s3_parquet_io_manager(init_context):
    return S3ParquetIOManager(
        bucket=init_context.resource_config["bucket"],
        prefix=init_context.resource_config.get("prefix", "data"),
    )

# Assign IO manager per asset
@dg.asset(io_manager_key="s3_parquet")
def silver_orders() -> pd.DataFrame:
    ...

defs = dg.Definitions(
    assets=[silver_orders],
    resources={
        "s3_parquet": s3_parquet_io_manager.configured(
            {"bucket": "data-lake", "prefix": "silver"}
        ),
    },
)

Built-in IO Managers

Sensors

Sensors poll external state and trigger runs.

Basic Sensor

@dg.sensor(
    target=dg.AssetSelection.assets(daily_events),
    minimum_interval_seconds=60,
    default_status=dg.DefaultSensorStatus.RUNNING,
)
def new_files_sensor(context: dg.SensorEvaluationContext):
    import boto3
    s3 = boto3.client("s3")

    last_seen = context.cursor or ""
    response = s3.list_objects_v2(Bucket="data-lake", Prefix="raw/events/")
    new_files = [
        obj["Key"] for obj in response.get("Contents", [])
        if obj["Key"] > last_seen
    ]

    if not new_files:
        return dg.SkipReason("No new files found")

    run_requests = [
        dg.RunRequest(
            run_key=f"file-{file_key.replace('/', '-')}",   # idempotency key
            run_config={"ops": {"daily_events": {"config": {"file": file_key}}}},
        )
        for file_key in new_files
    ]

    context.update_cursor(sorted(new_files)[-1])
    return run_requests

Asset Sensor — React to Upstream Materialization

@dg.asset_sensor(
    asset_key=dg.AssetKey("raw_orders"),
    target=dg.AssetSelection.assets(stg_orders),
)
def raw_orders_sensor(context: dg.SensorEvaluationContext, asset_event):
    return dg.RunRequest(
        run_key=f"orders-{asset_event.run_id}",
        tags={"triggered_by": "raw_orders_sensor"},
    )

Multi-Asset Sensor

@dg.multi_asset_sensor(
    monitored_assets=[dg.AssetKey("raw_orders"), dg.AssetKey("raw_customers")],
    target=dg.AssetSelection.assets(fct_orders),
)
def both_sources_sensor(context: dg.MultiAssetSensorEvaluationContext):
    latest_orders   = context.latest_materialization_records_by_key().get(dg.AssetKey("raw_orders"))
    latest_customers = context.latest_materialization_records_by_key().get(dg.AssetKey("raw_customers"))

    if latest_orders and latest_customers:
        context.advance_all_cursors()
        return dg.RunRequest(run_key=f"{latest_orders.run_id}-{latest_customers.run_id}")
    return dg.SkipReason("Waiting for both raw_orders and raw_customers")

Declarative Automation

Declarative automation replaces imperative sensors/schedules for common patterns.

@dg.asset(
    partitions_def=dg.DailyPartitionsDefinition("2024-01-01"),
    # Materialize every day at 3am UTC after upstream is ready
    automation_condition=dg.AutomationCondition.on_cron("0 3 * * *"),
)
def daily_report(...): ...

@dg.asset(
    # Eagerly re-materialize whenever any upstream changes
    automation_condition=dg.AutomationCondition.eager(),
)
def derived_table(stg_orders: pd.DataFrame): ...

@dg.asset(
    # Fill in any missing partitions when upstream is available
    automation_condition=dg.AutomationCondition.on_missing(),
)
def backfillable_asset(...): ...

Enable in Definitions:

defs = dg.Definitions(
    assets=all_assets,
    sensors=[dg.AutomationConditionSensorDefinition(
        name="automation_condition_sensor",
        target="*",
    )],
)

Jobs and Schedules

# Define a job from a subset of assets
etl_job = dg.define_asset_job(
    name="daily_etl_job",
    selection=dg.AssetSelection.groups("silver", "gold"),
    tags={"team": "data-platform"},
    config=dg.RunConfig(
        ops={"daily_events": dg.OpConfig(config={"timeout_seconds": 3600})}
    ),
)

# Daily schedule
daily_schedule = dg.ScheduleDefinition(
    job=etl_job,
    cron_schedule="0 4 * * *",       # 4am UTC daily
    execution_timezone="UTC",
    default_status=dg.DefaultScheduleStatus.RUNNING,
)

Project Structure

my_dagster_project/
├── my_dagster_project/
│   ├── __init__.py             # exports Definitions
│   ├── assets/
│   │   ├── __init__.py         # collect all assets
│   │   ├── bronze.py
│   │   ├── silver.py
│   │   └── gold.py
│   ├── resources/
│   │   ├── __init__.py
│   │   ├── warehouse.py        # WarehouseResource
│   │   └── io_managers.py      # S3ParquetIOManager
│   ├── sensors/
│   │   └── file_sensors.py
│   ├── jobs.py
│   └── definitions.py          # top-level Definitions object
├── tests/
│   └── test_assets.py
├── pyproject.toml              # [tool.dagster] section
└── dagster.yaml                # deployment config

# my_dagster_project/definitions.py
import dagster as dg
from my_dagster_project.assets import bronze, silver, gold
from my_dagster_project.resources import warehouse_resource, s3_parquet_io_manager
from my_dagster_project.sensors import new_files_sensor
from my_dagster_project.jobs import daily_schedule

defs = dg.Definitions(
    assets=dg.load_assets_from_modules([bronze, silver, gold]),
    resources={
        "warehouse": warehouse_resource,
        "io_manager": s3_parquet_io_manager,
    },
    sensors=[new_files_sensor],
    schedules=[daily_schedule],
)

Testing

# tests/test_assets.py
from dagster import materialize, build_sensor_context
from my_dagster_project.assets.silver import customers
from my_dagster_project.resources import WarehouseResource

def test_customers_asset():
    result = materialize(
        assets=[customers],
        resources={
            "warehouse": WarehouseResource(
                host="localhost",
                database="test_db",
                user="test",
                password="test",
            )
        },
    )
    assert result.success
    output = result.output_for_node("customers")
    assert len(output) > 0
    assert "customer_id" in output.columns
    assert output["email"].str.islower().all()

def test_sensor_yields_run_requests():
    context = build_sensor_context()
    result = new_files_sensor(context)
    # verify SkipReason on empty bucket
    assert isinstance(result, dg.SkipReason)

Materializing in Unit Tests

from dagster import materialize_to_memory

def test_pipeline():
    result = materialize_to_memory([raw_orders, stg_orders, fct_orders])
    assert result.success
    orders_df = result.output_for_node("fct_orders")
    assert "revenue_usd" in orders_df.columns

Deployment — Docker Compose

version: "3.8"
services:
  postgresql:
    image: postgres:14
    environment:
      POSTGRES_DB: dagster
      POSTGRES_USER: dagster
      POSTGRES_PASSWORD: dagster
    volumes:
      - dagster-db:/var/lib/postgresql/data

  dagster-webserver:
    image: my-dagster-project:latest
    command: dagster-webserver -h 0.0.0.0 -p 3000 -w workspace.yaml
    ports:
      - "3000:3000"
    environment:
      DAGSTER_POSTGRES_HOST: postgresql
      DAGSTER_POSTGRES_USER: dagster
      DAGSTER_POSTGRES_PASSWORD: dagster
      DAGSTER_POSTGRES_DB: dagster
    depends_on: [postgresql, dagster-daemon]

  dagster-daemon:
    image: my-dagster-project:latest
    command: dagster-daemon run
    environment:
      DAGSTER_POSTGRES_HOST: postgresql
      DAGSTER_POSTGRES_USER: dagster
      DAGSTER_POSTGRES_PASSWORD: dagster
      DAGSTER_POSTGRES_DB: dagster
    depends_on: [postgresql]

volumes:
  dagster-db:

# dagster.yaml
storage:
  postgres:
    postgres_db:
      username: dagster
      password: dagster
      hostname: postgresql
      db_name: dagster
      port: 5432

run_coordinator:
  module: dagster.core.run_coordinator
  class: QueuedRunCoordinator
  config:
    max_concurrent_runs: 10

Anti-Patterns

1. Putting business logic in sensors — sensors should only detect events and emit RunRequest. Move transformation logic into assets.

2. Missing run_key in sensors — without run_key, the same event triggers duplicate runs after restarts. Always set run_key to a unique, deterministic identifier.

3. Giant monolithic assets — one asset that loads, transforms, and writes is hard to retry and debug. Break into at least 3 layers: ingestion → staging → marts.

4. Not using resources for external connections — hardcoding connection strings in asset functions makes testing impossible. Always inject connections via ConfigurableResource.

5. Ignoring IO managers for large DataFrames — default pickle IO manager stores in memory and local disk. For production data volumes, use S3/GCS Parquet or warehouse IO managers.

6. Missing partition window checks — time-partitioned assets that don't filter by context.partition_time_window reprocess all historical data on every run.

7. Creating >100k partitions — Dagster UI degrades with very large partition sets. For high-cardinality partitions, use dynamic partitions and only add active ones.

8. Loading entire upstream assets as function parameters — for large tables, this loads the full DataFrame into memory. Use resources to query with partition predicates instead.

9. Not setting default_status=RUNNING on sensors — sensors start as STOPPED by default and won't run until manually enabled. Set DefaultSensorStatus.RUNNING for sensors that should always be active.

10. Using @op and @job instead of @asset — new projects should default to assets. The @op/@job API is lower-level and loses the lineage, metadata, and declarative automation benefits.

References to Consult When Needed

• Software-Defined Assets: docs.dagster.io/concepts/assets/software-defined-assets
• Partitioning: docs.dagster.io/guides/build/partitions-and-backfills/partitioning-assets
• IO Managers: docs.dagster.io/guides/build/io-managers
• Sensors: docs.dagster.io/guides/automate/sensors
• Declarative Automation: docs.dagster.io/guides/automate/declarative-automation
• Resources: docs.dagster.io/guides/build/resources