Dagster

Overview

Dagster organizes data pipelines around software-defined assets — declarations of the data artifacts your pipeline produces. Assets track lineage, enable incremental computation, and integrate with the Dagster UI. Version 1.13 (current as of October 2026) introduces Components, a scaffolding layer on top of the same asset/job/resource APIs described below; new projects created with the dg CLI use that layout, but the classic dagster CLI and plain Python definitions in this skill still work unchanged and are the simpler starting point for a single-file pipeline.

Instructions

Installation

# Install Dagster and the UI webserver
pip install dagster dagster-webserver

# Scaffold a project (classic layout: one code location, pyproject.toml, tests)
dagster project scaffold --name my_pipeline
cd my_pipeline
pip install -e ".[dev]"

# Start the dev server (loads Definitions objects found in the package)
dagster dev
# UI at http://localhost:3000

Newer projects can instead use uvx create-dagster@latest project my_pipeline (or pip install dagster-dg-cli then create-dagster project my_pipeline), which scaffolds a Components-ready layout and starts the UI with dg dev instead of dagster dev. Both layouts share the same dagster Python package and the asset/resource/schedule APIs below.

Software-Defined Assets

# my_pipeline/assets.py: Define assets that produce data
import dagster as dg
import pandas as pd
import httpx

@dg.asset(group_name="raw")
def raw_users(context: dg.AssetExecutionContext) -> pd.DataFrame:
    """Fetch raw user data from the billing API."""
    response = httpx.get("https://billing.internal.sharpline.io/v1/users")
    df = pd.DataFrame(response.json())
    context.log.info(f"Fetched {len(df)} users")
    return df

@dg.asset(group_name="raw")
def raw_orders(context: dg.AssetExecutionContext) -> pd.DataFrame:
    """Fetch raw order data from the billing API."""
    response = httpx.get("https://billing.internal.sharpline.io/v1/orders")
    return pd.DataFrame(response.json())

@dg.asset(group_name="analytics", deps=[raw_users, raw_orders])
def revenue_by_user(raw_users: pd.DataFrame, raw_orders: pd.DataFrame) -> pd.DataFrame:
    """Calculate total revenue per user."""
    merged = raw_orders.merge(raw_users, left_on="user_id", right_on="id")
    result = (
        merged.groupby(["user_id", "name"])
        .agg(total_revenue=("amount", "sum"), order_count=("id_x", "count"))
        .reset_index()
    )
    return result

dagster as dg is the import convention used throughout current Dagster docs and examples (dg.asset, dg.Definitions, …) — it is unrelated to the separate dg CLI installed by dagster-dg-cli. The older from dagster import asset, AssetExecutionContext style still works; both resolve to the same package.

Resources

# my_pipeline/resources.py: Configurable resources for external systems
import dagster as dg
import sqlalchemy

class DatabaseResource(dg.ConfigurableResource):
    connection_string: str

    def query(self, sql: str) -> list:
        engine = sqlalchemy.create_engine(self.connection_string)
        with engine.connect() as conn:
            result = conn.execute(sqlalchemy.text(sql))
            return [dict(row._mapping) for row in result]

    def execute(self, sql: str):
        engine = sqlalchemy.create_engine(self.connection_string)
        with engine.connect() as conn:
            conn.execute(sqlalchemy.text(sql))
            conn.commit()

Assets with Resources

# my_pipeline/db_assets.py: Assets that use database resources
import dagster as dg
from .resources import DatabaseResource

@dg.asset(group_name="warehouse")
def dim_users(context: dg.AssetExecutionContext, database: DatabaseResource):
    """Load cleaned user dimension table into warehouse."""
    users = database.query("SELECT id, name, email, created_at FROM raw_users")
    context.log.info(f"Loaded {len(users)} users into warehouse")
    return users

Definitions

# my_pipeline/__init__.py: Wire everything together
import dagster as dg
from . import assets, db_assets
from .resources import DatabaseResource
import os

all_assets = dg.load_assets_from_modules([assets, db_assets])

defs = dg.Definitions(
    assets=all_assets,
    resources={
        "database": DatabaseResource(
            connection_string=os.environ["WAREHOUSE_CONNECTION_STRING"]
        ),
    },
)

Schedules and Sensors

# my_pipeline/schedules.py: Time-based and event-based triggers
import os
import dagster as dg

# Job that materializes specific assets
analytics_job = dg.define_asset_job(
    name="analytics_job",
    selection=dg.AssetSelection.groups("analytics"),
)

# Cron schedule
daily_analytics = dg.ScheduleDefinition(
    job=analytics_job,
    cron_schedule="0 6 * * *",  # 6 AM daily
)

# Sensor — trigger on external event
@dg.sensor(job=analytics_job, minimum_interval_seconds=60)
def new_file_sensor(context: dg.SensorEvaluationContext):
    incoming_dir = os.environ["INCOMING_DATA_DIR"]
    new_files = [f for f in os.listdir(incoming_dir) if f.endswith(".csv")]
    if new_files:
        context.log.info(f"Found {len(new_files)} new files")
        yield dg.RunRequest(run_key=new_files[0])
    else:
        yield dg.SkipReason("No new files found")

Partitioned Assets

# my_pipeline/partitioned.py: Time-partitioned assets for incremental processing
import dagster as dg

daily_partitions = dg.DailyPartitionsDefinition(start_date="2026-01-01")

@dg.asset(partitions_def=daily_partitions, group_name="raw")
def daily_events(context: dg.AssetExecutionContext):
    """Fetch events for a specific date partition."""
    date = context.partition_key  # e.g., "2026-09-19"
    context.log.info(f"Processing events for {date}")
    return fetch_events(date)

CLI Reference

# Development server (classic layout); Components layout uses `dg dev` instead
dagster dev

# Materialize assets
dagster asset materialize --select raw_users,raw_orders

# List assets
dagster asset list

# Run a job
dagster job execute -j analytics_job

# Load and validate all Definitions without running anything
dagster definitions validate

Examples

Example 1: Build a daily revenue pipeline

Request: "Pull users and orders from our billing API every morning and compute revenue per user."

Write raw_users, raw_orders and revenue_by_user as shown in Software-Defined Assets above, wire them into a Definitions object, then schedule them:

import dagster as dg
from .assets import raw_users, raw_orders, revenue_by_user

analytics_job = dg.define_asset_job(name="analytics_job", selection=dg.AssetSelection.all())
daily_schedule = dg.ScheduleDefinition(job=analytics_job, cron_schedule="0 6 * * *")

defs = dg.Definitions(
    assets=[raw_users, raw_orders, revenue_by_user],
    schedules=[daily_schedule],
)

Result: dagster dev shows a three-node asset graph in the UI; the schedule runs at 06:00 daily, and dagster asset materialize --select raw_users,raw_orders,revenue_by_user runs it on demand. Each run's logs and the resulting revenue_by_user table are visible in the Runs tab.

Example 2: React to new files landing in a directory

Request: "A vendor drops CSV exports into a shared folder a few times a day. Kick off the import job automatically when a new file shows up."

Use the new_file_sensor from Schedules and Sensors above, pointed at the drop directory via INCOMING_DATA_DIR:

export INCOMING_DATA_DIR="/srv/dagster/incoming"
dagster dev

Result: the sensor polls every 60 seconds; when a .csv file appears, it yields a RunRequest keyed on the filename, so the same file never triggers two runs even if the sensor evaluates again before the run starts. When the folder is empty it yields SkipReason("No new files found"), which shows up in the sensor's tick history instead of a silent no-op.

Guidelines

• dagster dev is for local development: it runs an in-memory, single-process webserver and scheduler. For production, deploy with Dagster+ (the managed offering) or self-host dagster-webserver plus a daemon process behind a real run launcher and database — do not run dagster dev as a production service.
• Keep API calls, file I/O and database writes inside assets or resources, not at import time; Dagster imports your modules to build the asset graph, so top-level side effects run on every CLI invocation and UI reload.
• Prefer ConfigurableResource over module-level clients so connection strings and credentials come from Definitions(resources=...) (backed by environment variables), not hardcoded values — this also lets tests swap in a fake resource.
• Partition definitions must line up with the data: a DailyPartitionsDefinition asset downstream of a non-partitioned asset needs an explicit partition mapping, or materialization fails.
• dagster definitions validate loads every asset, job, schedule and sensor without executing them — run it in CI to catch import errors and bad resource wiring before deploying.
• Sensors poll; minimum_interval_seconds is a floor, not a guarantee — a slow sensor function or a busy instance can delay the next tick.
• For a new project today, create-dagster project (the dg CLI) is the path the official docs lead with and scaffolds Components support; dagster project scaffold still produces a working classic project and needs no extra packages, which is simpler for a single pipeline like the one above.