Dagster — Asset-First Orchestration · Data Engineering Quest

What changes when you think in assets

Dagster (1.10+ as of April 2026) is the orchestrator that asks "what data assets do you produce?" instead of "what tasks do you run?" An asset is a logical data object — a table, a Parquet directory, an ML model, a report — that gets materialized by a function. The DAG falls out of the dependency graph between assets, automatically.

This sounds like a small shift. In practice it changes the conversation. "Orders is stale because the upstream customers asset hasn't refreshed since Tuesday" is a sentence Dagster's UI lets you say; in Airflow you'd be looking at task statuses and inferring what they mean for the data.

The shape of a Dagster project

Assets — Python functions decorated with @asset. Each one returns the materialized object (or writes it via an IO manager).
IO Managers — pluggable strategies for how assets are read and written (Parquet on S3, Postgres, in-memory, etc).
Resources — reusable config (database connections, API clients) injected into assets.
Sensors / schedules — when assets should be re-materialized.

Code

A Dagster pipeline expressed as three assets·python

import pandas as pd
from dagster import asset, AssetExecutionContext

@asset(group_name='orders')
def raw_orders(context: AssetExecutionContext) -> pd.DataFrame:
    df = pd.read_csv('source/orders.csv', dtype=str)
    context.add_output_metadata({'rows': len(df)})
    return df

@asset(group_name='orders')
def clean_orders(context: AssetExecutionContext, raw_orders: pd.DataFrame) -> pd.DataFrame:
    df = (raw_orders
        .assign(order_date=lambda d: pd.to_datetime(d['order_date']),
                amount_usd=lambda d: pd.to_numeric(d['amount_usd']))
        .dropna(subset=['order_id', 'amount_usd']))
    context.add_output_metadata({'rows': len(df), 'dropped': len(raw_orders) - len(df)})
    return df

@asset(group_name='orders')
def monthly_revenue(clean_orders: pd.DataFrame) -> pd.DataFrame:
    return (clean_orders
        .assign(month=lambda d: d['order_date'].dt.to_period('M'))
        .groupby('month', as_index=False)['amount_usd'].sum()
        .rename(columns={'amount_usd': 'revenue'}))

Dagster — Asset-First Orchestration

What changes when you think in assets

The shape of a Dagster project

Code

External links

Exercise

Progress

Comments 0