Dagster: Data Orchestration with Assets, Jobs & Sensors

Dagster is the cloud-native data orchestration platform that treats data assets — not tasks — as first-class citizens. Get full data lineage, asset materialization tracking, and a rich observability UI out of the box.

What is Dagster?

Dagster is an open-source data orchestration platform designed around the concept of Software-Defined Assets (SDAs) — Python functions that declare what data artifacts they produce and depend on. Unlike Airflow which orchestrates tasks, Dagster orchestrates data: it tracks the materialization state of every asset, shows the full lineage graph from raw sources to final models, and surfaces data freshness and quality at the platform level.

Dagster integrates natively with dbt (model assets), Airbyte (source assets), Spark, Snowflake, BigQuery, and most of the modern data stack. The Dagit web UI provides an interactive asset graph, run history with per-step logs, a partition viewer for backfill monitoring, and a sensor dashboard. Dagster Cloud is the managed offering; Dagster Open Source can be self-hosted with dagster dev for local development.

Software-Defined Assets

# pip install dagster dagster-webserver
from dagster import asset, AssetExecutionContext, Definitions
import pandas as pd

# Software-Defined Assets — define the data, not the task
@asset(
    group_name="raw",
    description="Raw orders from the API",
)
def raw_orders(context: AssetExecutionContext) -> pd.DataFrame:
    import httpx
    data = httpx.get("https://api.example.com/orders").json()
    df = pd.DataFrame(data)
    context.log.info(f"Fetched {len(df)} orders")
    return df

@asset(
    group_name="staging",
    deps=[raw_orders],
    description="Cleaned orders with USD amounts",
)
def stg_orders(raw_orders: pd.DataFrame) -> pd.DataFrame:
    df = raw_orders.copy()
    df["amount_usd"] = df["amount_cents"] / 100
    df = df[df["status"] != "cancelled"]
    df["created_at"] = pd.to_datetime(df["created_at"])
    return df

@asset(
    group_name="marts",
    deps=[stg_orders],
    description="Revenue aggregated by region and month",
)
def revenue_by_region(stg_orders: pd.DataFrame) -> pd.DataFrame:
    return (
        stg_orders
        .assign(month=lambda df: df["created_at"].dt.to_period("M"))
        .groupby(["region", "month"])
        .agg(revenue=("amount_usd", "sum"), orders=("order_id", "count"))
        .reset_index()
    )

# Definitions — registers all assets, jobs, schedules, resources
defs = Definitions(assets=[raw_orders, stg_orders, revenue_by_region])

Jobs, Schedules, Sensors & Resources

# Jobs, Schedules, Sensors & Resources
from dagster import (
    define_asset_job, ScheduleDefinition, sensor,
    RunRequest, SensorEvaluationContext, resource,
    Definitions, EnvVar,
)
from dagster_postgres import PostgresIOManager

# Job — subset of assets to materialize together
etl_job = define_asset_job(
    name="nightly_etl",
    selection=["raw_orders", "stg_orders", "revenue_by_region"],
)

# Schedule — run the job on a cron
nightly_schedule = ScheduleDefinition(
    job=etl_job,
    cron_schedule="0 2 * * *",  # 2am UTC daily
    execution_timezone="UTC",
)

# Sensor — trigger a run when a new S3 file appears
@sensor(job=etl_job, minimum_interval_seconds=60)
def new_file_sensor(context: SensorEvaluationContext):
    import boto3
    s3 = boto3.client("s3")
    paginator = s3.get_paginator("list_objects_v2")
    for page in paginator.paginate(Bucket="my-bucket", Prefix="inbox/"):
        for obj in page.get("Contents", []):
            key = obj["Key"]
            if key not in context.cursor:
                yield RunRequest(
                    run_key=key,
                    run_config={"ops": {"raw_orders": {"config": {"s3_key": key}}}},
                )

# Resource — shared database connection
postgres_resource = PostgresIOManager(
    host=EnvVar("POSTGRES_HOST"),
    database=EnvVar("POSTGRES_DB"),
)

defs = Definitions(
    assets=[raw_orders, stg_orders, revenue_by_region],
    jobs=[etl_job],
    schedules=[nightly_schedule],
    sensors=[new_file_sensor],
    resources={"io_manager": postgres_resource},
)

Partitioned Assets & Backfills

# Partitioned assets — process data day by day
from dagster import (
    asset, DailyPartitionsDefinition, AssetExecutionContext,
    build_asset_context, materialize,
)

daily_partitions = DailyPartitionsDefinition(start_date="2024-01-01")

@asset(
    partitions_def=daily_partitions,
    group_name="raw",
)
def daily_events(context: AssetExecutionContext) -> pd.DataFrame:
    date = context.partition_key  # "2024-01-15"
    url = f"https://api.example.com/events?date={date}"
    return pd.DataFrame(httpx.get(url).json())

@asset(
    partitions_def=daily_partitions,
    deps=[daily_events],
)
def daily_summary(daily_events: pd.DataFrame) -> dict:
    return {
        "total_events": len(daily_events),
        "unique_users": daily_events["user_id"].nunique(),
    }

# Backfill a date range from CLI:
# dagster asset backfill --select daily_events daily_summary \
#   --partition-range 2024-01-01...2024-01-31

Dagster vs Airflow vs Prefect