Dagster for Data Engineering: Software-Defined Assets, Partitions & Auto-Materialize

dagster is the pick-one orchestration decision that decides whether your data platform is a graph of tasks that must be run in order or a graph of data that must exist and stay fresh — and it is the choice senior data engineers in 2026 get wrong most often because "we already have Airflow" is not the same argument as "Airflow is the right tool for our workload." Every dbt model, every S3 partition, every warehouse mart, every ML training feature your platform ships has to be scheduled, observed, backfilled, and quality-checked, and the mental model your orchestrator forces on you determines whether that observability is native or bolted on after the fact. The engineering trade-off does not live in "should we run an orchestrator" — every non-trivial data platform needs one — but in whether the platform describes itself as assets or as tasks, because that choice binds every downstream contract for years.

This guide is the senior-DE walkthrough you wished existed the first time an interviewer asked "walk me through Dagster's dagster software defined assets model and why it's structurally different from an Airflow DAG," or "what does dagster auto materialize actually replace, and why did the team rewrite it as Declarative Automation in 1.6?", or "how would you migrate a 400-DAG Airflow deployment onto Dagster without a big-bang cutover?". It walks through the five pieces of the Dagster mental model that senior interviewers score highest on — the asset-centric abstraction (@asset decorator, asset key hierarchy, asset checks, IO managers), the partitions v2 system (DailyPartitionsDefinition, dynamic partitions, multi-dimensional partitions, partition mappings), the new Declarative Automation DSL (AutomationCondition.on_cron(), on_missing(), any_deps_updated(), all_deps_updated()), the deep dagster-dbt integration that turns every dbt model into a first-class Dagster asset, and the dagster vs airflow migration story that lets you run both side by side while cutting over one team at a time. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the SQL practice library →, rehearse system design against the design practice library →, practice the aggregation shapes that dominate mart-level assets with the aggregation practice library →, and sharpen the streaming axis with the streaming practice library →.

---

On this page

• Why Dagster matters in 2026
• Software-Defined Assets and the asset graph
• Partitions, backfills and branch deployments
• Auto-Materialize and Declarative Automation
• Dagster vs Airflow migration and interview signals
• Cheat sheet — Dagster recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why Dagster matters in 2026

An asset-centric orchestrator was the missing piece for modern data platforms — Dagster is the reference implementation

The one-sentence invariant: dagster is an asset-centric orchestrator where the unit of work is a software-defined asset — a durable data object with a typed identity, upstream/downstream lineage, quality checks, and a materialization function — and the runtime, not the author, computes which assets need to run to bring the asset graph into a desired state; this mental-model flip from task-centric Airflow (where you schedule computations) to asset-centric Dagster (where you declare data that should exist) is the single largest reason senior data engineers pick Dagster for greenfield platforms in 2026, and the reason every migration answer starts with "yes, but which model does your team think in?". The mental model you pick in year one becomes the mental model every dbt model, every dashboard, every ML feature hard-codes assumptions against — so the decision matters far beyond "which scheduler do we install."

The four axes interviewers actually probe.

• Mental model. Airflow says "run these tasks in this order." Dagster says "these datasets should exist; here is how each one is built; here are its dependencies." The runtime computes the graph. Interviewers open here because it separates people who understand that orchestration and data modelling are the same job from people who've only maintained an Airflow scheduler.
• Data quality integration. Airflow treats data quality as an out-of-band step (a GreatExpectationsOperator after your compute task). Dagster treats it as a first-class asset construct via @asset_check — the check runs inside the same materialization event, blocks downstream materializations on failure by policy, and shows up in the same lineage UI as the asset itself. That's a structural difference, not a plugin choice.
• Partitions and backfills. Airflow has execution_date and catchup=True, but the partition abstraction is per-DAG and stringly-typed. Dagster's PartitionsDefinition is a first-class typed object that every downstream asset can reason about, and a backfill is a single UI action that spans the entire partition range across the entire asset graph. Multi-dimensional partitions (region × day) are native.
• Automation model. Airflow schedules tasks on cron. Dagster 1.6+ Declarative Automation says "materialize this asset when this condition is true" — on_cron('0 * * * *') & any_deps_updated(). The runtime evaluates the condition per asset and materializes only what needs to run. This is the single feature interviewers ask about most in 2026 because it replaced the older AutoMaterializePolicy in a controlled deprecation.

The 2026 reality — Dagster is one of four viable orchestrators.

• Dagster. Asset-centric, Python-native, comes with a great UI, ships with dagster-dbt deep integration and dagster-embedded-elt. Dagster+ (2024) is the managed offering with branch deployments, hosted Insights, and single-tenant compute. Adopted by teams that treat data-engineering-as-software-engineering and are willing to invest in an asset-first mental model.
• Airflow. Task-centric, Python-native, largest ecosystem of operators by an order of magnitude (500+ community + provider packages), massive install base. The default answer for teams whose primary need is "run this operator on this cron," or for teams whose existing platform is 90% Airflow already.
• Prefect. Task-centric with a lighter footprint and a strong "just write Python" story; smaller ecosystem than Airflow, less-featured data model than Dagster; a good fit for teams that want an orchestrator that stays out of the way.
• Argo Workflows / Kubernetes-native alternatives. Container-first, YAML-first, cloud-native; adopted by platform teams that want scheduling to look like Kubernetes primitives; typically paired with a separate data-catalog for the missing lineage story.

What interviewers listen for.

• Do you name "software-defined asset" as the Dagster unit of work, not "Dagster task"? — required answer.
• Do you distinguish asset-centric vs task-centric in the first sentence when comparing to Airflow? — senior signal.
• Do you name Declarative Automation and AutomationCondition as the 1.6+ replacement for AutoMaterializePolicy? — senior signal.
• Do you name dagster-dbt as deep integration (every dbt model = a Dagster asset), not just "there's an operator"? — required answer.
• Do you describe the orchestrator's job as "keeping the asset graph in a desired state," not "running tasks on a schedule"? — senior signal.

Worked example — the four-orchestrator comparison table

Detailed explanation. The single most useful artifact for a Dagster interview is a memorised 4x4 comparison table. Every senior orchestration discussion converges on this table within the first ten minutes; having it in your head is what separates a fluent answer from a stumbling one. Walk through building the table for a hypothetical greenfield analytics platform that needs to serve dbt marts, ML feature engineering, and a Kafka streaming sink.

• Workload. Analytics platform on Snowflake. ~200 dbt models, ~40 Python-based ML features, one streaming ingestion pipeline off Kafka. ~15 data engineers across three teams.
• Requirements. Native dbt integration, per-partition backfill, data-quality checks inline with materialization, a UI that shows lineage without a third-party catalog, and PR-scoped test deployments.
• Constraints. No existing orchestrator (greenfield), Python-first shop, Kubernetes-based compute, 24/7 on-call rotation of three engineers.

Question. Build the four-orchestrator comparison for the greenfield analytics platform and pick the tool each requirement drives you toward.

Input.

Orchestrator	Mental model	dbt integration	Backfills	UI + lineage
Dagster	asset-centric	dagster-dbt deep (model = asset)	first-class partitions v2	native asset graph + Insights
Airflow	task-centric	DbtOperator (run as task)	catchup + backfill CLI	task graph + external catalog
Prefect	task-centric	prefect-dbt operator	flow-level re-runs	task DAG UI
Argo	container-first	container image	manual per-workflow	template UI + external catalog

Code.

# Dagster: three lines to define a dbt-loaded asset graph
from dagster import Definitions
from dagster_dbt import DbtCliResource, dbt_assets

@dbt_assets(manifest="dbt_project/target/manifest.json")
def analytics_dbt_assets(context, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()

defs = Definitions(
    assets=[analytics_dbt_assets],
    resources={"dbt": DbtCliResource(project_dir="dbt_project")},
)

Step-by-step explanation.

1. The analytics platform needs dbt models to be first-class objects in the orchestrator UI — not tasks that happen to run dbt. Dagster's dagster-dbt package parses the dbt manifest.json and creates one Dagster asset per dbt model, complete with lineage, config schema, tags, and materialization events. Airflow's DbtOperator runs dbt as a Bash-style step; you never see individual models.
2. For per-partition backfill, Dagster's PartitionsDefinition (attached to the asset via partitions_def=DailyPartitionsDefinition(...)) turns any asset into a partitioned dataset. A backfill is a UI action that spans the range; Airflow requires catchup=True + airflow dags backfill with per-DAG semantics.
3. For inline data quality, Dagster's @asset_check runs in the same materialization event as the asset. On failure, the check emits a AssetCheckResult(passed=False) that the UI surfaces alongside the materialization event; downstream policies can be configured to block on it. Airflow requires a separate task + XCom + a Great Expectations DAG.
4. For the UI + lineage, Dagster ships a native asset-graph view that shows every asset, its dependencies, its most recent materialization, its check status, and its next scheduled run. Airflow ships a task-graph view that shows tasks; you install a separate catalog (DataHub, OpenMetadata) for asset-level lineage.
5. For PR-scoped test deployments, Dagster+ branch deployments provision a full copy of your definitions per PR (using the same code location loader) with an isolated Postgres/S3 namespace. Airflow-native equivalents require a full separate Airflow deployment per PR.

Output.

Requirement	Winner	Why
Native dbt	Dagster	manifest.json parse → 1 asset per model
Per-partition backfill	Dagster	PartitionsDefinition + backfill UI
Inline data checks	Dagster	@asset_check in same event
Native lineage UI	Dagster	asset graph is the primary view
PR-scoped deploys	Dagster+	branch deployments
Existing operator ecosystem	Airflow	500+ providers, wins by volume

Rule of thumb. Never pick an orchestrator based on "which one is trendy." Pick it based on (mental model × dbt integration × partition semantics × UI lineage) — the four axes. If you cannot answer "what is the unit of work?" in one word (task, asset, container, flow), you don't have enough constraints yet.

Worked example — what interviewers actually probe

Detailed explanation. The senior data-engineering orchestration interview has a predictable structure: the interviewer opens with an ambiguous question ("how would you schedule this pipeline?"), then progressively narrows to test whether you know the axes. The candidates who name the mental model in sentence one score highest; the candidates who describe "a cron job" score lowest. Walk through the interview grading rubric.

• Ambiguous opener. "How would you orchestrate this data platform?" — invites you to name a mental model.
• Follow-up 1. "Why Dagster over Airflow?" — probes the asset-vs-task axis.
• Follow-up 2. "How does Dagster handle partitions?" — probes partition sophistication.
• Follow-up 3. "What replaced AutoMaterializePolicy?" — probes 1.6+ Declarative Automation awareness.
• Follow-up 4. "How would you migrate from Airflow without a big bang?" — probes migration realism.

Question. Draft a 5-minute senior Dagster answer that covers all four axes without waiting to be asked.

Input.

Interview signal	Weak answer	Senior answer
Unit of work	"Dagster runs Python tasks"	"software-defined asset — data object with lineage + checks"
Partitions	"Dagster has partitions like Airflow"	"PartitionsDefinition is typed; multi-dimensional native"
Automation	"there's an auto-materialize thing"	"Declarative Automation with AutomationCondition DSL"
dbt	"there's a dbt operator"	"dagster-dbt parses manifest.json → 1 asset per model"
Migration	"rewrite everything"	"run both, migrate team-by-team via dagster-airflow"

Code.

Senior Dagster answer template (5 minutes)
==========================================

Minute 1 — name the mental model up front
  "Dagster is asset-centric. The unit of work is a software-defined
   asset — a data object with lineage, checks, and a materialization
   function. Airflow is task-centric; you schedule computations. That
   mental-model flip is the biggest reason we'd pick Dagster."

Minute 2 — partitions
  "PartitionsDefinition attaches to an asset. Daily, hourly, static,
   dynamic, multi-dimensional (region x day). A backfill is a single
   UI action across the asset graph. Partition mappings between
   assets (identity, last-N, all-upstream) are typed."

Minute 3 — automation
  "1.6+ shipped Declarative Automation with the AutomationCondition
   DSL — on_cron() & any_deps_updated() as one expression. It
   replaces the older AutoMaterializePolicy, which is now in
   deprecation. The runtime evaluates the condition per asset every
   ~30 seconds and materialises only what needs to run."

Minute 4 — dbt integration
  "dagster-dbt parses the dbt manifest.json and creates one Dagster
   asset per dbt model — lineage, config, tags, and materialisation
   events flow through. The dbt DAG becomes a first-class subgraph
   of the Dagster asset graph. Airflow's DbtOperator runs dbt as a
   task; you don't see individual models."

Minute 5 — migration + Airflow reality
  "For an existing Airflow shop we'd run both in parallel via
   `dagster-airflow` (Dagster asset that shells out to Airflow) or
   Astronomer's Cosmos-style integration. Migrate one team at a
   time; keep both sources of truth until the last DAG is off
   Airflow. Airflow still wins on operator-library breadth; be
   honest about that in the answer."

Step-by-step explanation.

1. Minute 1 is the crucial framing. Naming the mental model immediately — "asset-centric" versus "task-centric" — signals you're a decision-maker, not a task-runner. Weak candidates dive into tools ("we'd use KubernetesPodOperator…") before naming the model.
2. Minute 2 addresses the partition axis before the interviewer asks. Naming typed partitions and multi-dimensional partitions in one breath preempts the follow-up. Weak answers say "Dagster has partitions like Airflow"; they're not equivalent.
3. Minute 3 covers automation. Naming Declarative Automation and the DSL by name (on_cron & any_deps_updated) is a strong senior signal because it shows you've followed the 1.6+ release notes, not just the tutorial.
4. Minute 4 addresses dbt. The word "deep" is the shibboleth — "deep integration" versus "there's an operator." Naming manifest.json parsing shows you understand the mechanism, not just the marketing.
5. Minute 5 is the migration reality. Being honest that Airflow still wins on ecosystem breadth is a senior signal — refusing to concede any weakness in your preferred tool is a junior signal. Naming dagster-airflow (the parallel-run bridge) preempts the "big bang or nothing" trap.

Output.

Grading criterion	Weak score	Senior score
Names mental model in minute 1	rare	mandatory
Names typed partitions	occasional	required
Names Declarative Automation	rare	senior signal
Names dagster-dbt deep integration	rare	required
Names Airflow's ecosystem win	rare	senior signal

Rule of thumb. The senior Dagster answer is a 5-minute monologue that covers mental model, partitions, automation, dbt, and migration reality without waiting for the follow-ups. Rehearse it once; deploy it every time.

Worked example — the "pick the orchestrator" decision tree

Detailed explanation. Given a new platform, the senior architect runs a 4-question decision tree in their head. Codifying the tree makes the interview answer reproducible: any interviewer can hand you a scenario and you can walk the tree out loud. Walk through the tree with three canonical scenarios: greenfield analytics platform, brownfield Airflow shop, and a Kubernetes-native platform team.

• Q1. Is the unit of work naturally a dataset (dbt model, S3 partition, warehouse mart)? → yes = asset-centric (Dagster); no = go to Q2.
• Q2. Is the unit of work a task with a fixed operator (S3ToRedshiftOperator, EmailOperator)? → yes = Airflow; no = go to Q3.
• Q3. Is the primary compute a container running on Kubernetes with no data-lineage requirement? → yes = Argo; no = go to Q4.
• Q4. Do you want the lightest possible "just Python" orchestrator? → yes = Prefect; else default back to Dagster.

Question. Walk the decision tree for the three scenarios and record the orchestrator each ends up with.

Input.

Scenario	Q1 (dataset unit?)	Q2 (task+operator?)	Q3 (K8s container?)	Q4 (light Python?)
Greenfield analytics	yes	—	—	—
Brownfield 400-DAG Airflow	maybe	yes	—	—
Kubernetes ML training	no	no	yes	—

Code.

# Decision-tree helper (illustrative)
def pick_orchestrator(dataset_unit: bool,
                       task_operator: bool,
                       k8s_container: bool,
                       light_python: bool) -> str:
    """Return the primary orchestrator for a new platform."""
    if dataset_unit:
        return "dagster"
    if task_operator:
        return "airflow"
    if k8s_container:
        return "argo"
    if light_python:
        return "prefect"
    return "dagster"   # sensible default for modern data platforms


# Walk the three scenarios
print(pick_orchestrator(True,  False, False, False))
# → dagster

print(pick_orchestrator(False, True,  False, False))
# → airflow

print(pick_orchestrator(False, False, True,  False))
# → argo

Step-by-step explanation.

1. Scenario 1 — greenfield analytics platform with dbt at the core. The unit of work is naturally a model (a dataset). Q1 short-circuits to Dagster. This is the modern default for analytics-heavy teams.
2. Scenario 2 — brownfield Airflow shop with 400 DAGs of operator-based tasks. The unit of work is a task using a fixed operator. Q2 = yes → Airflow. Migrating to Dagster is a valid future project but not the answer to "what should we run today?".
3. Scenario 3 — Kubernetes-native ML training platform. The unit of work is a container, not a dataset or a task. Q3 = yes → Argo. Bolting Dagster onto a container-first platform is possible but adds accidental complexity.
4. The parallel branch is scenario 2 → "should we migrate to Dagster?" — the honest answer is "yes, over 12-18 months, team-by-team via dagster-airflow." Migration is a real project with real cost; being clear-eyed about that is a senior signal.
5. If none of Q1-Q4 pass, the default is Dagster because the asset model has the most surface for future data-quality and lineage work. Refuse to pick an orchestrator until the team can name the unit of work.

Output.

Scenario	Primary orchestrator	Migration path
Greenfield analytics	Dagster	—
Brownfield 400-DAG Airflow	Airflow (today), Dagster (12-18m)	dagster-airflow parallel
Kubernetes ML training	Argo	none

Rule of thumb. The four-question decision tree is a whiteboard-friendly answer. Practice walking it end-to-end so an interviewer can hand you any scenario and get an orchestrator name in under 60 seconds.

Senior interview question on orchestrator selection

A senior interviewer often opens with: "You are the tech lead for a 20-engineer analytics platform running dbt against Snowflake, plus five Python-based feature-engineering pipelines and one Kafka streaming ingest. You currently have no orchestrator (everything is on cron). Walk me through why you'd pick Dagster, the initial Definitions file, the first three assets, the first partition definition, and the first Declarative Automation condition."

Solution Using a Definitions module, dbt asset loader, a partitioned S3 asset, and a single AutomationCondition

# definitions.py — the top-level Dagster Definitions module
from dagster import (
    Definitions,
    AssetSelection,
    DailyPartitionsDefinition,
    AutomationCondition,
    define_asset_job,
    ScheduleDefinition,
    EnvVar,
)
from dagster_dbt import DbtCliResource, dbt_assets
from dagster_aws.s3 import S3Resource
from dagster_snowflake import SnowflakeResource

# 1. Load every dbt model as a first-class Dagster asset
@dbt_assets(manifest="dbt_project/target/manifest.json")
def analytics_dbt_assets(context, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()


# 2. A partitioned S3 raw-ingest asset (daily partitions)
daily_partitions = DailyPartitionsDefinition(start_date="2026-01-01")

from dagster import asset

@asset(
    partitions_def=daily_partitions,
    automation_condition=AutomationCondition.on_cron("15 * * * *"),
    group_name="raw",
    compute_kind="python",
)
def raw_orders(context, s3: S3Resource):
    """Land one day's orders JSONL into s3://acme/raw/orders/date=YYYY-MM-DD/."""
    day = context.partition_key
    key = f"raw/orders/date={day}/orders.jsonl"
    # (S3 put + source fetch elided)
    context.log.info(f"materialised raw_orders[{day}] to s3://acme/{key}")


# 3. A downstream partitioned aggregate — Dagster infers the graph edge
@asset(
    partitions_def=daily_partitions,
    deps=[raw_orders],
    automation_condition=AutomationCondition.eager(),
    group_name="clean",
    compute_kind="snowflake",
)
def orders_clean(context, snowflake: SnowflakeResource):
    """MERGE raw_orders[day] into analytics.orders_clean."""
    day = context.partition_key
    with snowflake.get_connection() as conn:
        conn.cursor().execute(f"""
            MERGE INTO analytics.orders_clean tgt
            USING (SELECT * FROM raw.orders WHERE order_date = '{day}') src
            ON tgt.order_id = src.order_id
            WHEN MATCHED     THEN UPDATE SET status = src.status, updated_at = current_timestamp()
            WHEN NOT MATCHED THEN INSERT (order_id, customer_id, total_cents, order_date)
                                   VALUES (src.order_id, src.customer_id, src.total_cents, src.order_date)
        """)


# 4. Wire it all together
defs = Definitions(
    assets=[analytics_dbt_assets, raw_orders, orders_clean],
    resources={
        "dbt":       DbtCliResource(project_dir="dbt_project"),
        "s3":        S3Resource(region_name="us-east-1"),
        "snowflake": SnowflakeResource(
            account=EnvVar("SNOWFLAKE_ACCOUNT"),
            user=EnvVar("SNOWFLAKE_USER"),
            password=EnvVar("SNOWFLAKE_PASSWORD"),
            database="ANALYTICS",
            warehouse="ETL_WH",
        ),
    },
    jobs=[
        define_asset_job(
            name="daily_analytics_refresh",
            selection=AssetSelection.groups("raw", "clean"),
            partitions_def=daily_partitions,
        )
    ],
    schedules=[
        ScheduleDefinition(
            name="daily_at_2am_utc",
            cron_schedule="0 2 * * *",
            job_name="daily_analytics_refresh",
        ),
    ],
)

Step-by-step trace.

Step	Value	Reasoning
Unit of work	@asset (data object)	asset-centric, not task-centric
dbt integration	@dbt_assets(manifest=...)	1 dagster asset per dbt model
Partitions	DailyPartitionsDefinition	typed; downstream inherits
Automation	AutomationCondition.on_cron & eager	Declarative Automation DSL
Resources	Definitions(resources=...)	dependency-injected, testable
Schedule + job	ScheduleDefinition + define_asset_job	classic cron as backup

After deployment, Dagster's UI shows one asset graph with three layers: raw_orders (partitioned by day, cron-driven every 15 minutes), orders_clean (partitioned by day, eagerly materialised after raw_orders), and the dbt subgraph (parsed from manifest.json) hanging downstream of orders_clean. The platform team writes zero task-graph YAML; the runtime computes what to run every 30 seconds.

Output:

Metric	Value
Lines of definitions code	~60 (all-in)
Assets managed	200 dbt + 2 python = 202
Partition definitions	1 (DailyPartitions)
Automation conditions	2 (on_cron + eager)
Backfill mechanism	UI action; per-partition retries
Data-quality primitive	@asset_check (see H2-2)

Why this works — concept by concept:

• Software-defined assets — the single Dagster primitive: a decorator over a Python function that declares the asset's identity, dependencies, partitions, and automation. Every downstream feature (lineage, backfill, checks, IO managers) hangs off this one construct.
• @dbt_assets deep integration — dagster-dbt parses dbt_project/target/manifest.json at load time and produces one Dagster asset per dbt model with correct lineage. The dbt DAG is not "run as a task"; it is the asset subgraph. This is the "deep" in "deep integration."
• Typed partitions — DailyPartitionsDefinition is a first-class object, not a string. Downstream assets that share the same partitions definition automatically inherit the partition mapping (identity); different partitions definitions can be bridged with explicit PartitionMapping classes.
• Declarative Automation — AutomationCondition.on_cron(...) and AutomationCondition.eager() are two conditions the runtime evaluates every ~30 seconds. On truthy, the runtime schedules a materialization for that asset partition. This replaces the older AutoMaterializePolicy (deprecated in 1.6).
• Cost — one Definitions module (~60 lines for 202 assets), one dbt manifest parse per code-location load (~seconds), one Postgres for the Dagster instance, and the runtime CPU cost of the condition evaluator (negligible for hundreds of assets). Compared to a cron-based platform, this buys lineage, backfill, and data-quality primitives at essentially zero incremental cost. O(assets) per graph load, O(1) per materialization decision.

SQL
Topic — sql
SQL practice for asset-driven mart queries

Practice →

Design Topic — design Design problems on modern orchestration

Practice →

---

2. Software-Defined Assets and the asset graph

dagster software defined assets are the unit of work — a data object with lineage, checks, and a materialization function

The mental model in one line: a dagster software-defined asset (SDA) is a Python function decorated with @asset that declares (a) the asset's identity via an AssetKey (defaults to the function name; nestable for hierarchies), (b) its upstream dependencies via typed function parameters or an explicit deps=[...] list, (c) how to materialise the underlying data (the function body), (d) optionally its partitions, checks, IO manager, group, compute kind, tags, and automation condition — and the Dagster runtime uses that metadata to build the full asset graph, decide what to run when a materialization is requested, surface lineage in the UI, and route inputs/outputs through configured IO managers. Every senior Dagster interview starts with this construct; nothing else makes sense without it.

The four dimensions every senior DE reasons about for an SDA.

• Identity. The AssetKey — a hierarchical name like AssetKey(["analytics", "sales", "orders_clean"]) — is the durable handle. Two assets with the same key are the same asset regardless of code location. Hierarchies are the way to model groups (a common namespace for a team) without stringy naming conventions.
• Dependencies. Declared implicitly by function parameter name (def orders_clean(raw_orders): ... = raw_orders is upstream) or explicitly via deps=[AssetKey(...), other_asset_fn]. Both compile to the same graph edge; use parameters when you want the value passed via IO manager, use deps= for pure dependency wiring.
• Materialization function. The Python function body. Called by the runtime when a materialization is requested. Receives an AssetExecutionContext with the partition key, log helpers, retry state, and access to resources. Returns a MaterializeResult (or an asset value that the IO manager persists).
• Metadata knobs. partitions_def, automation_condition, group_name, compute_kind, tags, owners, code_version, io_manager_key, check_specs. Every knob is optional; the minimal @asset is one line.

The asset key hierarchy — namespaces without string convention.

• Flat key. @asset def orders_clean(): ... produces AssetKey(["orders_clean"]).
• Prefixed key. @asset(key_prefix=["analytics", "sales"]) def orders_clean(): ... produces AssetKey(["analytics", "sales", "orders_clean"]).
• Group vs prefix. group_name is a UI-level grouping ("show me all sales assets"); key_prefix is a hard identity segment. Use both; they're orthogonal.
• Cross-code-location references. Two Dagster code locations can define assets that reference each other by AssetKey. This is how you build large multi-team deployments without one monolithic Definitions module.

Asset checks — data quality inside the orchestrator.

• What they are. A @asset_check decorated Python function that emits an AssetCheckResult(passed=bool, metadata={...}) during or after the asset's materialization. The runtime records the check result alongside the materialization event; the UI shows a green/red dot per check per asset.
• Blocking vs non-blocking. By default checks are non-blocking (informational). Configure blocking=True to fail the parent materialization if the check fails. Downstream automation policies can also be told to wait for check success.
• Where they run. In the same process as the asset materialization when defined on the asset (check_specs=[...]), or as separate runs when defined via @asset_check. Either way, the runtime holds the check result in the same event log.
• What to check. Row-count guardrails, freshness (max updated_at), null-ratio on required columns, referential integrity between assets, business-rule invariants (revenue >= 0).

IO managers — the "how do we read/write the actual bytes" layer.

• What they are. A pluggable class that knows how to load_input(...) and handle_output(...) for an asset's typed inputs and outputs. Dagster ships S3PickleIOManager, SnowflakePandasIOManager, DuckDBIOManager, DbioManager, and more; custom IO managers are ~30 lines.
• The default. In-memory (development). In production you almost always override to S3, Snowflake, or a data warehouse.
• Why they matter. Without an IO manager the asset function has to write to storage manually and downstream assets can't receive the value as a typed argument. With an IO manager, def orders_clean(raw_orders: pd.DataFrame) -> pd.DataFrame: transparently loads and stores DataFrames through S3 Parquet.
• Per-asset override. Set io_manager_key="snowflake_io_manager" on @asset to route just that asset through a specific IO manager; useful when different asset types live in different storage systems.

Common interview probes on SDAs.

• "What is a software-defined asset?" — required answer: a Python function decorated with @asset that declares identity, deps, materialization, and metadata.
• "How is @asset different from @op?" — @op is a task-graph primitive (older; still supported for imperative workflows); @asset is the data-graph primitive.
• "How do asset checks differ from tests?" — checks run at materialization time on production data; tests run at CI time on fixtures. Both matter.
• "What is an IO manager?" — the load-input/handle-output plugin that decouples the asset function from the storage layer.

Worked example — a three-asset pipeline with lineage, checks, and IO managers

Detailed explanation. The canonical minimal Dagster asset pipeline: raw_orders reads from an S3 landing zone; orders_clean transforms with pandas; daily_revenue aggregates for the dashboard. Each asset uses an IO manager so the runtime passes DataFrames through without the asset code doing S3 I/O. orders_clean carries a row-count asset check as a guardrail.

• Assets. raw_orders (source, partitioned by day), orders_clean (transform), daily_revenue (aggregate).
• IO manager. S3ParquetIOManager writes each partition to s3://acme/analytics/<asset>/date=YYYY-MM-DD/data.parquet.
• Asset check. orders_clean_row_count_check fails the materialization if row count is zero.

Question. Implement the three assets plus the check plus a shared IO manager, and show what Dagster's UI displays after one materialization.

Input.

Component	Value
Storage	S3 parquet
Frame type	pandas.DataFrame
Partition	DailyPartitionsDefinition
Check	row count > 0 on orders_clean

Code.

# assets.py — three-asset pipeline with checks + IO manager
import pandas as pd
from dagster import (
    asset,
    asset_check,
    AssetCheckResult,
    AssetCheckSpec,
    DailyPartitionsDefinition,
    MaterializeResult,
    MetadataValue,
)

daily = DailyPartitionsDefinition(start_date="2026-01-01")


@asset(
    partitions_def=daily,
    io_manager_key="s3_parquet_io",
    group_name="raw",
    compute_kind="s3",
    tags={"team": "analytics"},
)
def raw_orders(context) -> pd.DataFrame:
    """One day of raw orders as a DataFrame; IO manager writes to S3 parquet."""
    day = context.partition_key
    # In real life: read from Kafka / RDS / vendor API. Here: synthesise.
    df = pd.DataFrame({
        "order_id":    range(1, 101),
        "customer_id": range(1, 101),
        "total_cents": [100 * i for i in range(1, 101)],
        "status":      ["pending"] * 100,
        "order_date":  [day] * 100,
    })
    context.log.info(f"raw_orders[{day}] materialised {len(df)} rows")
    return df


@asset(
    partitions_def=daily,
    io_manager_key="s3_parquet_io",
    group_name="clean",
    compute_kind="pandas",
    check_specs=[
        AssetCheckSpec(
            name="row_count_gt_zero",
            asset="orders_clean",
            blocking=True,   # fail materialization on check failure
        )
    ],
)
def orders_clean(context, raw_orders: pd.DataFrame) -> MaterializeResult:
    """Clean + type-cast; emit the row-count check inline."""
    df = raw_orders.copy()
    df["total_dollars"] = df["total_cents"] / 100.0
    df = df.drop(columns=["total_cents"])
    n = len(df)

    return MaterializeResult(
        value=df,
        metadata={
            "row_count": MetadataValue.int(n),
            "columns":   MetadataValue.json(list(df.columns)),
            "preview":   MetadataValue.md(df.head().to_markdown()),
        },
        check_results=[
            AssetCheckResult(
                passed=(n > 0),
                check_name="row_count_gt_zero",
                metadata={"row_count": MetadataValue.int(n)},
            )
        ],
    )


@asset(
    partitions_def=daily,
    io_manager_key="s3_parquet_io",
    group_name="mart",
    compute_kind="pandas",
    tags={"tier": "dashboard"},
)
def daily_revenue(context, orders_clean: pd.DataFrame) -> pd.DataFrame:
    """Per-day revenue aggregate; feeds the CEO dashboard."""
    agg = (
        orders_clean
        .groupby("order_date", as_index=False)
        .agg(total_revenue_dollars=("total_dollars", "sum"),
             order_count=("order_id", "count"))
    )
    context.log.info(f"daily_revenue[{context.partition_key}] rows={len(agg)}")
    return agg

# resources.py — S3 parquet IO manager (illustrative; ~30 lines)
import io
import boto3
import pandas as pd
from dagster import ConfigurableIOManager

class S3ParquetIOManager(ConfigurableIOManager):
    bucket: str
    prefix: str

    def _key(self, context) -> str:
        asset_name = context.asset_key.path[-1]
        partition = context.partition_key
        return f"{self.prefix}/{asset_name}/date={partition}/data.parquet"

    def handle_output(self, context, obj: pd.DataFrame) -> None:
        buf = io.BytesIO()
        obj.to_parquet(buf, index=False)
        boto3.client("s3").put_object(
            Bucket=self.bucket, Key=self._key(context), Body=buf.getvalue()
        )

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

Step-by-step explanation.

1. The three @asset functions form an implicit graph — Dagster infers raw_orders → orders_clean → daily_revenue from parameter names. No task-graph YAML, no set_upstream calls. The graph is the code.
2. Each asset shares partitions_def=daily — the identity partition mapping is implicit. Materialising daily_revenue[2026-07-15] triggers orders_clean[2026-07-15] triggers raw_orders[2026-07-15] in dependency order.
3. The io_manager_key="s3_parquet_io" on each asset routes the returned DataFrame through the S3ParquetIOManager. handle_output writes to s3://acme/analytics/<asset>/date=YYYY-MM-DD/data.parquet; load_input reads it back when a downstream asset needs it as a parameter. The asset code never touches S3.
4. The check_specs=[AssetCheckSpec(name="row_count_gt_zero", asset="orders_clean", blocking=True)] declares a check that runs inside the materialization event. MaterializeResult(check_results=[AssetCheckResult(passed=...)]) reports the outcome; blocking=True fails the materialization on passed=False, which cascades to prevent downstream assets from consuming stale/broken data.
5. MaterializeResult(metadata={...}) attaches per-materialization metadata that the UI renders inline: row count as an integer badge, the schema as a JSON block, the preview as a markdown table. This is how observability gets built into every asset — not as a separate ELT-QA step.

Output.

Asset	Materialization event	Check	UI badge
raw_orders[2026-07-15]	success, 100 rows	—	100 rows
orders_clean[2026-07-15]	success, 100 rows	row_count_gt_zero passed	100 rows, ✓ check
daily_revenue[2026-07-15]	success, 1 row	—	1 row

Rule of thumb. For any three-asset-or-larger pipeline, wire an IO manager on day one and add at least one AssetCheckSpec per critical asset. The check is the single cheapest correctness guardrail Dagster gives you; the IO manager is the single cheapest way to keep asset code storage-agnostic.

Worked example — asset checks for row-count and freshness guardrails

Detailed explanation. Every mart-tier asset in a serious production Dagster deployment needs at least two checks: a row-count guardrail (fail if the row count drops by more than N% versus the previous materialization) and a freshness check (fail if the source data is older than N minutes). Walk through both checks and the metadata payload the Dagster UI renders.

• Row-count guardrail. Compare len(df) against the previous materialization's row count (from the asset's own event log).
• Freshness check. Compare now() - max(updated_at) against an SLA of, say, 60 minutes.
• Both blocking. Fail the materialization on either.

Question. Write two @asset_check functions for the orders_clean asset — row count within 10% of the previous partition, and freshness within 60 minutes.

Input.

Check	Rule	Blocking?
row_count_within_10pct	abs(now - prev) / prev < 0.1	yes
freshness_lt_60min	now - max(updated_at) < 60 min	yes

Code.

# checks.py — two guardrails on the orders_clean asset
from datetime import datetime, timezone, timedelta

import pandas as pd
from dagster import (
    AssetCheckResult,
    AssetCheckSeverity,
    MetadataValue,
    asset_check,
    AssetKey,
)


@asset_check(
    asset=AssetKey(["orders_clean"]),
    blocking=True,
    description="Row count must be within 10% of the previous partition.",
)
def orders_clean_row_count_within_10pct(context) -> AssetCheckResult:
    """Compare this partition's row count against the last successful one."""
    now_rows = _load_row_count(context, "orders_clean", context.partition_key)
    prev_rows = _load_previous_row_count(context, "orders_clean", context.partition_key)

    if prev_rows == 0:
        return AssetCheckResult(
            passed=True,
            severity=AssetCheckSeverity.WARN,
            metadata={"note": MetadataValue.text("no previous partition to compare")},
        )

    drift = abs(now_rows - prev_rows) / prev_rows
    passed = drift < 0.10

    return AssetCheckResult(
        passed=passed,
        severity=AssetCheckSeverity.ERROR if not passed else AssetCheckSeverity.WARN,
        metadata={
            "now_rows":       MetadataValue.int(now_rows),
            "prev_rows":      MetadataValue.int(prev_rows),
            "drift_pct":      MetadataValue.float(round(drift * 100, 2)),
            "threshold_pct":  MetadataValue.float(10.0),
        },
    )


@asset_check(
    asset=AssetKey(["orders_clean"]),
    blocking=True,
    description="Max(updated_at) must be within 60 minutes of now.",
)
def orders_clean_freshness_lt_60min(context) -> AssetCheckResult:
    df: pd.DataFrame = _load_asset_partition(context, "orders_clean", context.partition_key)

    max_ts = pd.to_datetime(df["updated_at"]).max().to_pydatetime()
    if max_ts.tzinfo is None:
        max_ts = max_ts.replace(tzinfo=timezone.utc)
    age = datetime.now(timezone.utc) - max_ts
    passed = age < timedelta(minutes=60)

    return AssetCheckResult(
        passed=passed,
        metadata={
            "max_updated_at":  MetadataValue.text(max_ts.isoformat()),
            "age_seconds":     MetadataValue.int(int(age.total_seconds())),
            "threshold_min":   MetadataValue.int(60),
        },
    )


# Helpers (illustrative; real impl reads Dagster event log via instance client)
def _load_row_count(context, name: str, part: str) -> int:  # pragma: no cover
    ...

def _load_previous_row_count(context, name: str, part: str) -> int:  # pragma: no cover
    ...

def _load_asset_partition(context, name: str, part: str) -> pd.DataFrame:  # pragma: no cover
    ...

Step-by-step explanation.

1. @asset_check decorates a function that returns an AssetCheckResult. The asset= argument binds the check to a specific asset key; blocking=True tells the runtime to fail the parent materialization if the check fails; description= shows up in the UI tooltip.
2. The row-count check compares this partition's row count against the previous partition's row count (retrieved from the Dagster instance's event log via a helper). If drift exceeds 10%, the check fails; the parent materialization fails; downstream assets don't consume the suspicious data.
3. AssetCheckSeverity distinguishes WARN from ERROR; when blocking=True, only ERROR fails the materialization. Use WARN for informational drift (e.g., 5-9% drift is worth notice but not a page); use ERROR for hard limits.
4. The freshness check reads the actual asset's DataFrame (again via a helper — in production this would come from the IO manager) and computes now() - max(updated_at). If the source is stale, downstream mart materializations block. This is the check that catches "upstream job silently died at 3 AM."
5. The metadata dict is what the UI renders inline: now_rows, prev_rows, drift_pct, threshold_pct for the row check; max_updated_at, age_seconds, threshold_min for freshness. Every check emits enough context that on-call can triage without opening logs.

Output.

Check	Result	UI badge
orders_clean_row_count_within_10pct	passed, drift 2.1%	green ✓, "drift 2.1%"
orders_clean_freshness_lt_60min	passed, age 8 min	green ✓, "8m fresh"
(row check fails)	drift 47% → ERROR blocking	red ✗, materialization blocked
(freshness fails)	age 3h → ERROR blocking	red ✗, downstream halted

Rule of thumb. For any mart-tier or dashboard-feeding asset, ship at least one row-count-drift check and one freshness check with blocking=True. These two guardrails together catch 80% of the "we deployed at 3 AM and nobody noticed" incidents senior on-call rotations dread.

Worked example — IO manager routing across S3, Snowflake, and DuckDB

Detailed explanation. A realistic Dagster deployment often has multiple asset "tiers" that live in different storage systems: raw ingest lands in S3 parquet; cleaned intermediate lives in DuckDB for cheap local iteration; mart-tier lands in Snowflake for BI. Each tier wants a different IO manager. Walk through the config that routes assets to the right storage via io_manager_key.

• Raw tier. S3 parquet.
• Intermediate tier. DuckDB.
• Mart tier. Snowflake.
• Selection. io_manager_key="s3_parquet_io" per-asset, or per-group defaults.

Question. Wire three IO managers into the Definitions and route the three assets to the right one via per-asset io_manager_key.

Input.

Tier	IO manager	Storage location
raw	S3ParquetIOManager	s3://acme/raw/
clean	DuckDBIOManager	./duckdb/analytics.duckdb
mart	SnowflakeIOManager	analytics.mart schema

Code.

# definitions.py — multi-IO-manager routing
from dagster import Definitions, EnvVar
from dagster_snowflake_pandas import SnowflakePandasIOManager
from dagster_duckdb_pandas import DuckDBPandasIOManager

from .assets import raw_orders, orders_clean, daily_revenue
from .resources import S3ParquetIOManager


s3_parquet_io = S3ParquetIOManager(
    bucket="acme-lake",
    prefix="analytics/raw",
)

duckdb_io = DuckDBPandasIOManager(
    database="./duckdb/analytics.duckdb",
    schema="analytics",
)

snowflake_io = SnowflakePandasIOManager(
    account=EnvVar("SNOWFLAKE_ACCOUNT"),
    user=EnvVar("SNOWFLAKE_USER"),
    password=EnvVar("SNOWFLAKE_PASSWORD"),
    database="ANALYTICS",
    warehouse="ETL_WH",
    schema="MART",
)


defs = Definitions(
    assets=[raw_orders, orders_clean, daily_revenue],
    resources={
        "s3_parquet_io":  s3_parquet_io,   # raw_orders uses this
        "duckdb_io":      duckdb_io,       # orders_clean uses this
        "snowflake_io":   snowflake_io,    # daily_revenue uses this
    },
)

# assets.py — per-asset io_manager_key override
from dagster import asset, DailyPartitionsDefinition

daily = DailyPartitionsDefinition(start_date="2026-01-01")

@asset(partitions_def=daily, io_manager_key="s3_parquet_io")
def raw_orders(context): ...

@asset(partitions_def=daily, io_manager_key="duckdb_io")
def orders_clean(context, raw_orders): ...

@asset(partitions_def=daily, io_manager_key="snowflake_io")
def daily_revenue(context, orders_clean): ...

Step-by-step explanation.

1. Three IO managers are registered under three distinct keys in Definitions.resources. Each is a fully-configured, self-contained instance — the S3 one knows its bucket, the DuckDB one knows its database path, the Snowflake one knows its warehouse and schema.
2. Each @asset picks its IO manager via io_manager_key="...". The runtime uses this key to look up the resource in Definitions.resources at materialization time. Different assets use different IO managers; the same asset always uses the same IO manager unless you override at job-run time.
3. When Dagster materialises orders_clean, it needs the raw_orders value as a function argument. The runtime asks the upstream asset's IO manager (the S3 one) to load_input(...), gets back a DataFrame, and passes it to orders_clean. The runtime then asks the current asset's IO manager (DuckDB) to handle_output(...) on the return value.
4. This mechanism means each asset's storage is independent. Rewriting daily_revenue to land in BigQuery instead of Snowflake is a one-line change (io_manager_key="bigquery_io" + registering the resource). No asset code changes.
5. The IO manager contract is load_input(context) -> value and handle_output(context, value) -> None. Custom IO managers subclass ConfigurableIOManager; ~30 lines gets you a working parquet-on-S3 manager. This is the extension point that makes Dagster storage-agnostic.

Output.

Asset	IO manager	Storage after materialization
raw_orders[2026-07-15]	s3_parquet_io	s3://acme-lake/analytics/raw/raw_orders/date=2026-07-15/data.parquet
orders_clean[2026-07-15]	duckdb_io	./duckdb/analytics.duckdb → analytics.orders_clean, partition=2026-07-15
daily_revenue[2026-07-15]	snowflake_io	ANALYTICS.MART.daily_revenue, WHERE order_date='2026-07-15'

Rule of thumb. Choose one IO manager per storage system, register it under a descriptive key in Definitions.resources, and set io_manager_key per asset. Never mix "asset code writes to S3 directly" with "asset code returns a value" — pick one pattern per asset and stick to it.

Senior interview question on Software-Defined Assets

A senior interviewer might ask: "You have a three-tier analytics pipeline — raw S3 ingest, DuckDB intermediate, Snowflake mart. Design the Dagster asset graph, wire IO managers, add row-count and freshness checks on the mart-tier asset, and show what the UI displays after a full materialization plus one failed check."

Solution Using @asset, per-tier IO managers, and blocking asset checks

# definitions.py — the full SDA-first analytics pipeline
from dagster import (
    Definitions,
    EnvVar,
    asset,
    asset_check,
    AssetCheckResult,
    AssetCheckSpec,
    AssetKey,
    DailyPartitionsDefinition,
    MaterializeResult,
    MetadataValue,
)
from dagster_snowflake_pandas import SnowflakePandasIOManager
from dagster_duckdb_pandas import DuckDBPandasIOManager

import pandas as pd
from datetime import datetime, timezone, timedelta

daily = DailyPartitionsDefinition(start_date="2026-01-01")


# 1. Raw tier — S3 parquet
@asset(
    partitions_def=daily,
    io_manager_key="s3_parquet_io",
    group_name="raw",
    compute_kind="s3",
)
def raw_orders(context) -> pd.DataFrame:
    day = context.partition_key
    # fetch upstream, elided
    return pd.DataFrame({
        "order_id":   range(1, 1001),
        "customer_id": [i % 100 for i in range(1000)],
        "total_cents": [100 * (i % 500 + 1) for i in range(1000)],
        "order_date":  [day] * 1000,
        "updated_at":  [datetime.now(timezone.utc).isoformat()] * 1000,
    })


# 2. Clean tier — DuckDB
@asset(
    partitions_def=daily,
    io_manager_key="duckdb_io",
    group_name="clean",
    compute_kind="pandas",
)
def orders_clean(context, raw_orders: pd.DataFrame) -> pd.DataFrame:
    df = raw_orders.copy()
    df["total_dollars"] = df["total_cents"] / 100.0
    df = df[df["total_dollars"] > 0]
    return df.drop(columns=["total_cents"])


# 3. Mart tier — Snowflake, with two blocking checks
@asset(
    partitions_def=daily,
    io_manager_key="snowflake_io",
    group_name="mart",
    compute_kind="snowflake",
    check_specs=[
        AssetCheckSpec(name="row_count_gt_zero",  asset="daily_revenue", blocking=True),
        AssetCheckSpec(name="freshness_lt_60min", asset="daily_revenue", blocking=True),
    ],
)
def daily_revenue(context, orders_clean: pd.DataFrame) -> MaterializeResult:
    agg = (
        orders_clean
        .groupby("order_date", as_index=False)
        .agg(
            total_revenue_dollars=("total_dollars", "sum"),
            order_count=("order_id", "count"),
        )
    )
    n = len(agg)

    max_ts = pd.to_datetime(orders_clean["updated_at"]).max().to_pydatetime()
    if max_ts.tzinfo is None:
        max_ts = max_ts.replace(tzinfo=timezone.utc)
    age_sec = int((datetime.now(timezone.utc) - max_ts).total_seconds())

    return MaterializeResult(
        value=agg,
        metadata={
            "row_count":  MetadataValue.int(n),
            "revenue_sum": MetadataValue.float(float(agg["total_revenue_dollars"].sum())),
            "preview":    MetadataValue.md(agg.head().to_markdown()),
        },
        check_results=[
            AssetCheckResult(
                check_name="row_count_gt_zero",
                passed=(n > 0),
                metadata={"row_count": MetadataValue.int(n)},
            ),
            AssetCheckResult(
                check_name="freshness_lt_60min",
                passed=(age_sec < 3600),
                metadata={
                    "age_seconds":   MetadataValue.int(age_sec),
                    "threshold_sec": MetadataValue.int(3600),
                },
            ),
        ],
    )


defs = Definitions(
    assets=[raw_orders, orders_clean, daily_revenue],
    resources={
        "s3_parquet_io": ...,   # see H2-2 IO manager example
        "duckdb_io":     DuckDBPandasIOManager(
            database="./duckdb/analytics.duckdb", schema="analytics"
        ),
        "snowflake_io":  SnowflakePandasIOManager(
            account=EnvVar("SNOWFLAKE_ACCOUNT"),
            user=EnvVar("SNOWFLAKE_USER"),
            password=EnvVar("SNOWFLAKE_PASSWORD"),
            database="ANALYTICS",
            warehouse="ETL_WH",
            schema="MART",
        ),
    },
)

Step-by-step trace.

Layer	Component	Purpose
Raw	@asset raw_orders + s3_parquet_io	land day's rows to S3 parquet
Clean	@asset orders_clean + duckdb_io	typed transform in local DuckDB
Mart	@asset daily_revenue + snowflake_io	Snowflake table for BI
Checks	2x AssetCheckSpec on daily_revenue	row_count + freshness, both blocking
Metadata	MetadataValue.int/float/md	UI badges + inline preview
Definitions	resources={...}	one IO manager per tier

After deployment, materialising daily_revenue[2026-07-15] triggers a dependency-ordered run: raw_orders → orders_clean → daily_revenue. Each asset writes through its tier's IO manager. The two checks run inside the daily_revenue materialization and emit results; the UI shows two green dots. If either check fails, daily_revenue fails and downstream assets that depend on it don't materialise.

Output:

Metric	Value
Assets in graph	3
IO managers	3 (S3, DuckDB, Snowflake)
Checks	2 blocking on daily_revenue
Metadata per materialization	row count, revenue sum, preview
Storage per asset	independent per tier
Lineage in UI	native asset graph, 3-node

Why this works — concept by concept:

• Software-defined assets as the primitive — every entity in the graph is a data object with lineage, checks, and materialization semantics. Task-graph thinking is replaced by data-graph thinking; the runtime schedules the computations required to bring the data graph into the desired state.
• Per-tier IO managers — S3 for raw, DuckDB for intermediate, Snowflake for mart. The asset code returns a DataFrame; the IO manager handles the storage. Rewriting daily_revenue to land in BigQuery is a one-line resource change.
• Inline AssetCheckSpec + AssetCheckResult — the row-count and freshness checks run inside the mart-tier materialization event. blocking=True prevents downstream consumers from seeing broken data. The UI surfaces green/red dots per check per asset.
• Rich MetadataValue — every materialization attaches row counts, aggregates, and a markdown preview. This is the observability layer Dagster gets for free that Airflow requires XComs + a Grafana panel to approximate.
• Cost — three IO manager instantiations, one Definitions module, one Postgres for the Dagster instance, negligible per-materialization overhead. The eliminated cost is the ad-hoc "we bolted GreatExpectations onto Airflow" and "we wrote a lineage sidecar" scaffolding that most task-centric platforms accrete. O(assets) per graph load; O(1) per materialization decision; O(1) per check evaluation.

SQL
Topic — sql
SQL practice for mart-tier aggregate assets

Practice →

Aggregation Topic — aggregation Aggregation practice for asset graph outputs

Practice →

---

3. Partitions, backfills and branch deployments

dagster partitions are first-class typed objects — daily, hourly, static, dynamic, multi-dimensional — with per-partition backfill baked in

The mental model in one line: a dagster PartitionsDefinition is a typed object attached to an @asset that names the set of partition keys the asset is materialised against — DailyPartitionsDefinition(start_date=...) produces one key per calendar day, StaticPartitionsDefinition([...]) produces a fixed key set, DynamicPartitionsDefinition(name=...) grows keys at runtime, MultiPartitionsDefinition({"region": ..., "day": ...}) composes two dimensions — and the runtime uses these keys to schedule per-partition materializations, run backfills across partition ranges, and propagate mappings between upstream and downstream assets that don't share the same partitions definition. Every senior Dagster interview probes partitions because they are the abstraction that separates "nightly full refresh" from "one materialization per day per region per team."

The five kinds of partitions every senior DE names.

• Time-window (Daily/Hourly/Monthly). DailyPartitionsDefinition(start_date="2026-01-01"), HourlyPartitionsDefinition(start_date="2026-01-01-00:00"), MonthlyPartitionsDefinition(start_date="2026-01-01"). Each key is an ISO date/time string; the runtime computes elapsed keys from start_date to now. This is 80% of production partitions.
• Static. StaticPartitionsDefinition(["us-east", "us-west", "eu-central"]). A fixed list. Used for region-per-partition, team-per-partition, tenant-per-partition. Keys never change without a code deploy.
• Dynamic. DynamicPartitionsDefinition(name="customer_id"). Partition keys are added at runtime via context.instance.add_dynamic_partitions(...). Used when the partition set grows during operation (per-customer, per-experiment).
• Multi-dimensional. MultiPartitionsDefinition({"region": StaticPartitionsDefinition([...]), "day": DailyPartitionsDefinition(...)}). Every partition is a tuple of one key per dimension (e.g., {"region": "us-east", "day": "2026-07-15"}). This is the abstraction that makes region × day dashboards trivial.
• Custom. Subclass PartitionsDefinition when the built-in shapes don't fit. Rare; usually only for fiscal-calendar or business-week partition schemes.

Partition mappings — how upstream partitions map to downstream partitions.

• Identity (default when partitions definitions match). daily_revenue[2026-07-15] reads orders_clean[2026-07-15]. The runtime infers this when both assets share the same partitions_def.
• Time window (last N). LastPartitionMapping() — the downstream partition reads the most recent upstream partition. Used for "the current dashboard reads whichever is the latest ingest partition."
• All-upstream. AllPartitionMapping() — the downstream partition reads every upstream partition. Used for aggregations across all history (e.g., "lifetime revenue").
• Custom time-window. TimeWindowPartitionMapping(start_offset=-7, end_offset=0) — the downstream partition reads the last 8 upstream partitions. Used for 7-day rolling windows.
• Multi to single dimension. When a downstream asset is single-dim (day) and upstream is multi-dim (region x day), a MultiToSingleDimensionPartitionMapping fans in region.

Backfills — the "fill in the gap" operation.

• What. A backfill materialises a range of partitions for one or more assets in a single click / API call. Dagster+ makes it a UI action; OSS Dagster provides a CLI (dagster asset backfill).
• Single-run vs partition-by-partition. By default a backfill launches one run per partition (parallel up to your concurrency limit). For assets that support it (via backfill_policy=BackfillPolicy.single_run()), the runtime can launch one run that processes the whole range — much cheaper for warehouse SQL that supports WHERE day BETWEEN.
• Retries. A failed backfill partition is retried without restarting the whole backfill. The UI shows a partition grid with red/green per cell.
• Cost control. Backfills respect run-launcher concurrency limits and per-op tags, so you can --tag priority=low and the launcher schedules them behind priority=high runs.

Branch deployments — PR-scoped test infrastructure (Dagster+).

• What. Dagster+ provisions a full, isolated copy of your Definitions per PR — Postgres namespace, code location, run launcher, storage prefix. Assets materialise into s3://acme-lake/branch-deployments/pr-42/... rather than production storage.
• Why. Test that a new asset definition, partition scheme, or automation condition works end-to-end without polluting production data. Reviewers can click "materialise" in the PR-scoped UI and see real behaviour.
• Cost. One Dagster instance per PR (managed by Dagster+); short-lived (PR-scoped); expires when the PR closes.
• The alternative in OSS. A separate Dagster deployment per PR (feasible but manual); most teams settle for a single staging deployment.

Common interview probes on partitions.

• "How does Dagster handle partitions compared to Airflow?" — typed, per-asset, and multi-dimensional; Airflow's execution_date is a per-DAG string.
• "How would you backfill 30 days?" — UI action; per-partition runs by default; single-run policy for range-friendly assets.
• "How do you map partitions between assets with different definitions?" — PartitionMapping classes (Last, All, TimeWindow, MultiToSingleDimension).
• "Have you used branch deployments?" — Dagster+ PR-scoped stacks; explain the isolated namespace trade-off.

Worked example — hourly-partitioned S3 raw into daily-partitioned aggregate

Detailed explanation. The canonical time-window partition example: raw events land in S3 hourly (HourlyPartitionsDefinition), and a daily aggregate rolls up the 24 upstream hourly partitions into one downstream partition using a TimeWindowPartitionMapping. Walk through both partition definitions and the mapping.

• Raw. raw_events — hourly, one partition per hour.
• Daily. daily_events_agg — daily, one partition per day; each reads the 24 hourly partitions of that day.
• Mapping. TimeWindowPartitionMapping() between hourly and daily is Dagster's implicit default for calendar-aligned time windows.

Question. Write the two assets, register the partition mappings, and show what one daily materialization triggers upstream.

Input.

Asset	PartitionsDefinition	Cardinality
raw_events	HourlyPartitionsDefinition(start="2026-01-01-00:00")	24 partitions/day
daily_events_agg	DailyPartitionsDefinition(start="2026-01-01")	1 partition/day
Mapping	TimeWindowPartitionMapping (implicit)	24-to-1

Code.

# assets.py — hourly raw → daily aggregate
from dagster import (
    asset,
    HourlyPartitionsDefinition,
    DailyPartitionsDefinition,
    AssetIn,
    TimeWindowPartitionMapping,
    MaterializeResult,
    MetadataValue,
)
import pandas as pd

hourly = HourlyPartitionsDefinition(start_date="2026-01-01-00:00")
daily  = DailyPartitionsDefinition(start_date="2026-01-01")


@asset(
    partitions_def=hourly,
    io_manager_key="s3_parquet_io",
    group_name="raw",
    compute_kind="s3",
)
def raw_events(context) -> pd.DataFrame:
    """One hour of raw events; keyed by ISO hour string."""
    hour_key = context.partition_key   # e.g. "2026-07-15-14:00"
    n = 500
    return pd.DataFrame({
        "event_id":   range(n),
        "event_ts":   [hour_key] * n,
        "user_id":    [i % 100 for i in range(n)],
        "event_type": (["view", "click", "purchase"] * (n // 3 + 1))[:n],
    })


@asset(
    partitions_def=daily,
    io_manager_key="s3_parquet_io",
    group_name="agg",
    compute_kind="pandas",
    ins={
        "raw_events": AssetIn(
            partition_mapping=TimeWindowPartitionMapping()
        ),
    },
)
def daily_events_agg(context, raw_events: dict[str, pd.DataFrame]) -> MaterializeResult:
    """Roll up 24 hourly partitions into one daily aggregate.

    Because the upstream is hourly and this asset is daily, Dagster
    delivers `raw_events` as a dict[str, pd.DataFrame] keyed by the
    upstream hourly partition_key.
    """
    day_key = context.partition_key      # e.g. "2026-07-15"

    all_rows = pd.concat(raw_events.values(), ignore_index=True) \
                 if raw_events else pd.DataFrame()

    agg = (
        all_rows
        .groupby("event_type", as_index=False)
        .agg(n=("event_id", "count"))
    )
    agg["day"] = day_key

    return MaterializeResult(
        value=agg,
        metadata={
            "upstream_partitions_read": MetadataValue.int(len(raw_events)),
            "day":                      MetadataValue.text(day_key),
            "row_count":                MetadataValue.int(len(agg)),
        },
    )

Step-by-step explanation.

1. HourlyPartitionsDefinition(start_date="2026-01-01-00:00") produces one key per hour since the start date; today would give 24 * 200 partitions after 200 days. Each is an ISO string like "2026-07-15-14:00".
2. DailyPartitionsDefinition(start_date="2026-01-01") produces one key per day. daily_events_agg[2026-07-15] needs the 24 upstream raw_events partitions from 2026-07-15-00:00 through 2026-07-15-23:00.
3. AssetIn(partition_mapping=TimeWindowPartitionMapping()) tells Dagster that the calendar-aligned time-window mapping is intended — one downstream day maps to 24 upstream hours. Dagster resolves the upstream partition keys automatically.
4. Because the mapping is many-to-one, the runtime delivers raw_events as dict[str, pd.DataFrame] keyed by the upstream partition key, not as a single DataFrame. This is the typed contract for a fan-in.
5. pd.concat(raw_events.values()) flattens the 24 hourly DataFrames into one; the aggregation then produces one row per event_type for the day. The metadata surfaces upstream_partitions_read so on-call can spot "we only got 22 of 24 hours" without opening logs.

Output.

Layer	Partitions	Materialization pattern
raw_events	24 partitions per day	one run per hour
daily_events_agg[2026-07-15]	1 partition per day	one run reads 24 upstream
Delivery shape	dict[str, DataFrame]	keyed by hourly partition
Metadata badge	"upstream_partitions_read: 24"	catches missing hours

Rule of thumb. Whenever an aggregate asset has a coarser time grain than its upstream, declare the TimeWindowPartitionMapping() explicitly via AssetIn. The runtime infers reasonable mappings when both sides are time-window and calendar-aligned, but the explicit AssetIn is what makes the intent readable and grep-able for future maintainers.

Worked example — a 30-day backfill across the asset graph

Detailed explanation. The senior on-call runbook: raw data was late for 30 days; a bug in the source system meant raw_orders never materialised. Now the source is fixed and you need to backfill 30 days of raw_orders, cascading through orders_clean and daily_revenue. Dagster's backfill UI (or the CLI) makes this one command. Walk through it.

• Assets. raw_orders, orders_clean, daily_revenue (from H2-2), all daily-partitioned.
• Range. 2026-06-15 → 2026-07-14 (30 days).
• Concurrency. Cap at 5 parallel partitions to avoid saturating Snowflake.

Question. Kick off a 30-day backfill via the CLI, respect the concurrency cap, and show the partition grid Dagster displays.

Input.

Parameter	Value
Assets	raw_orders, orders_clean, daily_revenue
Start date	2026-06-15
End date	2026-07-14
Concurrency	5
Retry policy	3 retries with exponential backoff

Code.

# Backfill 30 days of the asset selection via CLI
dagster asset backfill \
    --select 'raw_orders* or orders_clean* or daily_revenue*' \
    --partitions '2026-06-15..2026-07-14' \
    --tags '{"priority": "backfill", "owner": "analytics"}'

# Cap concurrency in dagster.yaml
# run_coordinator:
#   module: dagster.core.run_coordinator
#   class:  QueuedRunCoordinator
#   config:
#     max_concurrent_runs: 5
#     tag_concurrency_limits:
#       - key: priority
#         value: backfill
#         limit: 5

# For range-friendly assets — single-run backfill policy
from dagster import asset, DailyPartitionsDefinition, BackfillPolicy

daily = DailyPartitionsDefinition(start_date="2026-01-01")


@asset(
    partitions_def=daily,
    io_manager_key="snowflake_io",
    backfill_policy=BackfillPolicy.single_run(),
    compute_kind="snowflake",
)
def daily_revenue_single_run(context) -> None:
    """Snowflake can compute 30 days in one MERGE; use single-run backfill."""
    keys = context.partition_key_range   # (start, end)
    with context.resources.snowflake.get_connection() as conn:
        conn.cursor().execute(f"""
            MERGE INTO analytics.mart.daily_revenue tgt
            USING (
                SELECT order_date,
                       SUM(total_dollars) AS total_revenue_dollars,
                       COUNT(*)           AS order_count
                FROM   analytics.clean.orders_clean
                WHERE  order_date BETWEEN '{keys.start}' AND '{keys.end}'
                GROUP  BY order_date
            ) src
            ON tgt.order_date = src.order_date
            WHEN MATCHED     THEN UPDATE SET
                tgt.total_revenue_dollars = src.total_revenue_dollars,
                tgt.order_count           = src.order_count
            WHEN NOT MATCHED THEN INSERT (order_date, total_revenue_dollars, order_count)
                                   VALUES (src.order_date, src.total_revenue_dollars, src.order_count)
        """)

Step-by-step explanation.

1. dagster asset backfill --select ... --partitions '2026-06-15..2026-07-14' launches a backfill for the asset selection across the partition range. Dagster resolves the 30 partition keys and creates 30 * 3 = 90 individual run entries (one per (asset, partition) pair in default per-partition mode).
2. The --tags '{"priority": "backfill"}' tag propagates to every run. The QueuedRunCoordinator in dagster.yaml uses tag_concurrency_limits to cap priority=backfill runs at 5 in parallel — the other 85 wait in the queue. This is how you keep the backfill from saturating Snowflake.
3. The UI's partition grid shows a 30-column × 3-row grid, one cell per (partition, asset). Cells go grey → yellow (running) → green (success) or red (failure) as runs progress. Retries update the cell colour in place; you don't need to re-launch the whole backfill.
4. For range-friendly assets like daily_revenue_single_run, BackfillPolicy.single_run() tells the runtime to materialise the whole range in one execution. context.partition_key_range gives a (start, end) tuple; the SQL uses WHERE ... BETWEEN to process everything in one MERGE. Cost drops from 30 Snowflake warehouses spinning up to 1.
5. Failed partitions get retried per the RetryPolicy (3 attempts with exponential backoff). Persistent failures stay red in the grid; the on-call opens the individual run to debug. When resolved, right-click → "Retry" on the red cell re-launches just that partition.

Output.

Metric	Per-partition	Single-run
Runs launched	30 * 3 = 90	3 (one per asset, whole range)
Concurrency cap effect	5 parallel; 90 total	irrelevant (1 run each)
Snowflake WH spin-ups	30	1 per asset
Retry granularity	per partition	per whole range
Best for	non-range compute (S3, HTTP)	warehouse SQL

Rule of thumb. For every asset with a partitions_def, decide up front whether BackfillPolicy.single_run() fits the compute. Warehouse SQL, dbt models, and any compute that supports WHERE partition BETWEEN should be single-run; per-partition-only computes (per-key HTTP fetches, per-day file writes) stay on the default per-partition policy.

Worked example — a branch deployment for a PR that adds a new asset

Detailed explanation. The Dagster+ branch-deployment workflow: a data engineer opens a PR that adds a new asset (daily_conversion_rate) reading daily_revenue. Dagster+ provisions a branch deployment tied to that PR — the new asset appears in an isolated UI, materialises into a PR-scoped S3 prefix, and the reviewer can click "materialise" to see it run end-to-end without touching production data. Walk through the config and the reviewer flow.

• PR branch. feature/conversion-rate-asset.
• New asset. daily_conversion_rate computed from daily_revenue.
• Storage isolation. s3://acme-lake/branch-deployments/pr-42/.
• Reviewer flow. Approve → materialise in PR UI → check row count → merge.

Question. Write the new asset and describe the Dagster+ branch-deployment behaviour when the PR is opened.

Input.

Component	Value
New asset	daily_conversion_rate
Upstream	daily_revenue
Branch deployment name	pr-42
Storage prefix	s3://acme-lake/branch-deployments/pr-42/

Code.

# assets.py — new asset added in the PR
from dagster import asset, DailyPartitionsDefinition, MaterializeResult, MetadataValue
import pandas as pd

daily = DailyPartitionsDefinition(start_date="2026-01-01")


@asset(
    partitions_def=daily,
    io_manager_key="s3_parquet_io",
    group_name="mart",
    compute_kind="pandas",
    tags={"team": "growth"},
)
def daily_conversion_rate(context, daily_revenue: pd.DataFrame) -> MaterializeResult:
    """Placeholder: conversion rate = order_count / visits (visits from another source)."""
    df = daily_revenue.copy()
    # In real life: join to visits asset. Here we fake it for the branch deploy.
    df["visits"]          = df["order_count"] * 25
    df["conversion_rate"] = df["order_count"] / df["visits"]

    return MaterializeResult(
        value=df,
        metadata={
            "row_count":       MetadataValue.int(len(df)),
            "mean_conv_rate":  MetadataValue.float(round(float(df["conversion_rate"].mean()), 4)),
            "preview":         MetadataValue.md(df.head().to_markdown()),
        },
    )

# dagster_cloud.yaml — how Dagster+ discovers this code location
locations:
  - location_name: analytics
    code_source:
      package_name: acme_analytics
    build:
      directory: ./

# Dagster+ branch-deployment lifecycle for PR #42
1. PR opened   → Dagster+ receives GitHub webhook
2. Provision   → new branch deployment `pr-42` created
3. Build       → dagster_cloud.yaml resolved; code location loaded
4. Isolate     → S3 prefix + Postgres namespace scoped to pr-42
5. UI link     → PR check "Dagster branch deployment ready" links to
                  https://acme.dagster.plus/deployments/pr-42
6. Materialise → reviewer clicks "materialise all" on pr-42 UI
7. Cleanup     → PR closed → deployment auto-torn-down after 24h

Step-by-step explanation.

1. The PR adds one asset (daily_conversion_rate) that depends on daily_revenue. No production code changes — just a new @asset definition. The PR passes CI locally and gets pushed.
2. Dagster+'s GitHub App receives the PR-opened webhook and provisions a branch deployment named pr-42. This is a full Dagster instance — new Postgres namespace, new S3 storage prefix, new run launcher — that mirrors production Definitions plus the PR's changes.
3. The PR gets a GitHub check "Dagster branch deployment ready" with a link to the pr-42 UI. Reviewers open it and see the full asset graph including the new asset. Materialising anything writes to s3://acme-lake/branch-deployments/pr-42/... — production storage is untouched.
4. The reviewer clicks "materialise" on daily_conversion_rate[2026-07-15]. Dagster runs the full dependency chain in the branch deployment (using upstream partitions from raw_orders, orders_clean, daily_revenue re-materialised into the pr-42 prefix). The reviewer sees the row count, the preview markdown, and the mean_conv_rate badge.
5. When the PR merges, the branch deployment is torn down after 24 hours (default TTL). If the PR closes without merging, the same TTL applies. Storage is garbage-collected automatically. The reviewer never has to think about "did I clean up my test data."

Output.

Lifecycle stage	Producer	Artifact
PR opened	GitHub webhook	branch deployment pr-42 provisioned
Build	Dagster+ builder	code location loaded from PR SHA
Isolation	Dagster+	Postgres namespace + S3 prefix scoped
Review	Human reviewer	click "materialise" in pr-42 UI
Merge	GitHub merge	branch deployment TTL starts
Cleanup	Dagster+	automatic teardown after 24h

Rule of thumb. For any team on Dagster+, use branch deployments as the default review artifact — no reviewer should approve a new asset PR without materialising it once in the PR-scoped UI. For OSS Dagster teams, the closest equivalent is a shared staging deployment where reviewers manually create a per-team asset selection.

Senior interview question on partitions

A senior interviewer might ask: "Design a multi-region analytics pipeline where raw events land per (region, hour) partition, a daily per-region aggregate rolls up 24 hours, and a global daily aggregate fans-in three regions. Wire the partition definitions, the partition mappings, and demonstrate how a 7-day backfill across all three assets is a single UI action."

Solution Using MultiPartitionsDefinition, TimeWindowPartitionMapping, and MultiToSingleDimensionPartitionMapping

# assets.py — multi-region pipeline with MultiPartitionsDefinition + mappings
from dagster import (
    asset,
    AssetIn,
    HourlyPartitionsDefinition,
    DailyPartitionsDefinition,
    StaticPartitionsDefinition,
    MultiPartitionsDefinition,
    TimeWindowPartitionMapping,
    MultiToSingleDimensionPartitionMapping,
    MaterializeResult,
    MetadataValue,
)
import pandas as pd

# 1. Partition definitions
regions = StaticPartitionsDefinition(["us-east", "us-west", "eu-central"])
hourly  = HourlyPartitionsDefinition(start_date="2026-01-01-00:00")
daily   = DailyPartitionsDefinition(start_date="2026-01-01")

# raw_events_regional is partitioned by (region, hour)
region_hourly = MultiPartitionsDefinition({
    "region": regions,
    "hour":   hourly,
})

# daily_regional_agg is partitioned by (region, day)
region_daily = MultiPartitionsDefinition({
    "region": regions,
    "day":    daily,
})


@asset(
    partitions_def=region_hourly,
    io_manager_key="s3_parquet_io",
    group_name="raw",
    compute_kind="s3",
)
def raw_events_regional(context) -> pd.DataFrame:
    keys = context.partition_key.keys_by_dimension
    region, hour = keys["region"], keys["hour"]
    return pd.DataFrame({
        "event_id":   range(500),
        "region":     [region] * 500,
        "event_ts":   [hour] * 500,
        "user_id":    [i % 100 for i in range(500)],
    })


@asset(
    partitions_def=region_daily,
    io_manager_key="s3_parquet_io",
    group_name="agg",
    compute_kind="pandas",
    ins={
        "raw_events_regional": AssetIn(
            partition_mapping=TimeWindowPartitionMapping(),  # 24 hours → 1 day, per region
        ),
    },
)
def daily_regional_agg(context, raw_events_regional: dict[str, pd.DataFrame]) -> MaterializeResult:
    keys = context.partition_key.keys_by_dimension
    region, day = keys["region"], keys["day"]
    all_rows = pd.concat(raw_events_regional.values(), ignore_index=True) \
                 if raw_events_regional else pd.DataFrame()
    n = len(all_rows)
    return MaterializeResult(
        value=all_rows,
        metadata={
            "region":      MetadataValue.text(region),
            "day":         MetadataValue.text(day),
            "upstream_hr": MetadataValue.int(len(raw_events_regional)),
            "row_count":   MetadataValue.int(n),
        },
    )


@asset(
    partitions_def=daily,
    io_manager_key="s3_parquet_io",
    group_name="mart",
    compute_kind="pandas",
    ins={
        "daily_regional_agg": AssetIn(
            partition_mapping=MultiToSingleDimensionPartitionMapping(
                partition_dimension_name="day"
            ),
        ),
    },
)
def daily_global_agg(context, daily_regional_agg: dict[str, pd.DataFrame]) -> MaterializeResult:
    """Fan in all three regions for the given day."""
    day = context.partition_key
    per_region = pd.concat(daily_regional_agg.values(), ignore_index=True) \
                    if daily_regional_agg else pd.DataFrame()
    n = len(per_region)
    return MaterializeResult(
        value=per_region,
        metadata={
            "day":              MetadataValue.text(day),
            "regions_read":     MetadataValue.int(len(daily_regional_agg)),
            "row_count_total":  MetadataValue.int(n),
        },
    )

# 7-day backfill across all three assets in one UI/CLI action
dagster asset backfill \
    --select 'raw_events_regional* or daily_regional_agg* or daily_global_agg*' \
    --partitions '2026-07-08..2026-07-14'

Step-by-step trace.

Layer	Component	Partitions	Materialization pattern
Raw	raw_events_regional	(region, hour) — MultiPartitionsDefinition	3 * 24 = 72 per day
Regional agg	daily_regional_agg	(region, day) — MultiPartitionsDefinition	3 per day
Global agg	daily_global_agg	day — DailyPartitionsDefinition	1 per day
Mapping raw → regional	TimeWindowPartitionMapping	24 hours → 1 day, per region	fan-in
Mapping regional → global	MultiToSingleDimensionPartitionMapping(day)	3 regions → 1 global	fan-in
Backfill	CLI: --partitions '2026-07-08..2026-07-14'	7 days	one command, whole graph

After deployment, the UI shows one asset graph with three layers plus a two-dimensional partition grid on each multi-partitioned asset. The 7-day backfill launches 7 * 72 + 7 * 3 + 7 = 574 partition-run entries (or fewer with single_run policies on range-friendly aggs). The concurrency limit throttles them; the partition grid fills in green as each cell completes.

Output:

Metric	Value
Partition definitions	3 (region, hourly, daily + composites)
Partition mappings	2 explicit (TimeWindow + MultiToSingle)
Runs per day (per-partition)	72 + 3 + 1 = 76
Runs per 7-day backfill	~574 (default policy)
UI actions to backfill	1 (CLI or button)
Partition-grid dimensions	2D on multi-partitioned assets

Why this works — concept by concept:

• MultiPartitionsDefinition — composes two named dimensions into tuple-valued partitions. Every senior "region × day" or "team × experiment" or "customer × month" workload uses this instead of stringly-concatenating.
• TimeWindowPartitionMapping — the 24-hourly-to-1-daily fan-in per region. Explicit AssetIn(partition_mapping=...) communicates the intent to Dagster and to future maintainers reading the code.
• MultiToSingleDimensionPartitionMapping — the 3-regions-to-1-global fan-in. partition_dimension_name="day" tells the mapping which dimension is preserved across the boundary. Fans-in the other dimension.
• Single backfill CLI — one command materialises 574 partition-run entries. The runtime handles retries, concurrency limits, and per-cell status in the UI. No custom "backfill script per team" scaffolding.
• Cost — three partition definitions (all typed), two explicit mappings (both typed), one Definitions module, one Postgres for the Dagster instance. The eliminated cost is the "per-team backfill script" and the ad-hoc "regional × daily" partition-string convention that most Airflow shops accrete. O(assets * partitions) per backfill; O(1) per partition-run scheduling decision. Compared to Airflow's catchup=True + per-DAG backfill CLI, this is one command versus one command per DAG per day.

Aggregation
Topic — aggregation
Aggregation practice for partitioned rollups

Practice →

Design Topic — design Design problems on partitioned pipelines

Practice →

---

4. Auto-Materialize and Declarative Automation

dagster auto materialize was the old policy engine — Declarative Automation (1.6+) is the new AutomationCondition DSL that replaces it

The mental model in one line: AutomationCondition is the composable DSL that Dagster 1.6+ ships as the replacement for AutoMaterializePolicy — you attach an AutomationCondition to an @asset (e.g. automation_condition=AutomationCondition.on_cron("0 * * * *") & AutomationCondition.any_deps_updated()), the runtime evaluates the condition per asset partition every ~30 seconds, and launches materializations for any (asset, partition) pair whose condition is truthy — so instead of writing schedules and sensors that decide when to run assets, you declare when the asset should be fresh and let the runtime compute the schedule. The old AutoMaterializePolicy (eager, lazy, freshness_policy) is now a legacy shim on top of the same evaluator; senior interviewers want to hear the new DSL.

The AutomationCondition atoms — the building blocks.

• on_cron(cron_expr). True on every cron tick. Replaces classic schedules for the "run this asset every hour" pattern.
• on_missing(). True when the asset partition has never been materialised. Used to auto-fill new partitions as the calendar advances.
• any_deps_updated(). True when any upstream asset partition has been materialised since this asset was last materialised. This is the "react to upstream" primitive.
• all_deps_updated(). True only when all upstream asset partitions have been materialised since this asset was last materialised. Stricter than any_deps_updated; used when the asset should wait for a complete fan-in.
• in_progress(). True while a materialization for this asset is already running. Typically negated (& ~AutomationCondition.in_progress()) to prevent double-launching.
• newly_updated(). True when the asset was materialised in the last tick.

Composing conditions — the boolean algebra.

• & (and). on_cron("0 * * * *") & all_deps_updated() — run hourly and wait for all upstreams.
• | (or). on_cron("0 * * * *") | on_missing() — run hourly or if the partition is missing.
• ~ (not). ~in_progress() — do not launch if a run is already going.
• .since_last_handled(). Modifier that scopes evaluation to time since the runtime last emitted a materialization request for this (asset, partition). Prevents spammy re-emission.

The three built-in policies as AutomationCondition shortcuts.

• AutomationCondition.eager(). Materialise as soon as any upstream is updated. Equivalent to on_missing() | (any_deps_updated() & ~in_progress()).
• AutomationCondition.on_cron(...). Run on a cron; will still respect ~in_progress under the hood.
• AutomationCondition.on_missing(). Only fill missing partitions; never re-run existing ones.

Sensors, schedules, and Declarative Automation composed.

• When to use a @schedule. Classic cron-driven fan-outs that don't fit the on_cron shortcut — e.g., you want to run a job (multiple assets in a fixed order) rather than let the runtime auto-choose.
• When to use a @sensor. Reacting to external events — a new file in S3, a message on SQS, a completion signal from an out-of-Dagster system. Sensors evaluate on a tick and emit RunRequest for the runtime to launch.
• When to use Declarative Automation. Everything else. The bulk of modern Dagster deployments run 80% of their scheduling via Declarative Automation and only fall back to @schedule / @sensor for cron-fan-outs and external event triggers.

Back-pressure and rate limiting.

• Concurrency limits. tag_concurrency_limits in dagster.yaml caps parallel runs by tag (e.g., max concurrent snowflake=5).
• Global cap. max_concurrent_runs in the QueuedRunCoordinator caps the whole deployment.
• Per-asset cap. AutomationCondition naturally rate-limits by evaluating per-asset — you can't spam more materializations than the tick interval allows.
• Custom backoff. A run-status @sensor can watch for repeated failures and pause an asset's automation condition by setting an AutomationConditionSensorDefinition override.

Common interview probes on Declarative Automation.

• "What replaced AutoMaterializePolicy?" — required answer: AutomationCondition in Dagster 1.6+.
• "Give me an example of a composed condition." — on_cron('0 * * * *') & any_deps_updated().
• "How does Declarative Automation differ from a @schedule?" — per-asset, per-partition, condition-driven; schedules run whole jobs on a fixed cron.
• "How do you rate-limit auto-materializations?" — tag_concurrency_limits + QueuedRunCoordinator; conditions naturally rate-limit via tick interval.

Worked example — an eager pipeline that self-materializes on upstream freshness

Detailed explanation. The canonical Declarative Automation setup: attach AutomationCondition.eager() to every downstream asset. Whenever raw_orders materialises, the runtime detects the upstream update and eagerly materialises orders_clean, then daily_revenue. No @schedule needed for the downstreams. Walk through the four-asset chain.

• Root asset. raw_orders — cron-driven (on_cron("0 * * * *")).
• Downstream assets. orders_clean, daily_revenue, weekly_revenue — all eager.
• Behaviour. Root materialises hourly; downstreams cascade automatically.

Question. Wire the four assets with the right automation_condition on each, and describe what the runtime does over an hour.

Input.

Asset	AutomationCondition
raw_orders	on_cron("0 * * * *")
orders_clean	eager()
daily_revenue	eager()
weekly_revenue	eager()

Code.

# assets.py — eager cascade under one on_cron root
from dagster import asset, AutomationCondition, DailyPartitionsDefinition
import pandas as pd

daily = DailyPartitionsDefinition(start_date="2026-01-01")

@asset(
    partitions_def=daily,
    automation_condition=AutomationCondition.on_cron("0 * * * *"),
    io_manager_key="s3_parquet_io",
    group_name="raw",
    compute_kind="s3",
)
def raw_orders(context) -> pd.DataFrame:
    """Hourly fetch of the source; runtime evaluates cron every ~30s."""
    ...


@asset(
    partitions_def=daily,
    automation_condition=AutomationCondition.eager(),
    io_manager_key="s3_parquet_io",
    group_name="clean",
    compute_kind="pandas",
)
def orders_clean(context, raw_orders: pd.DataFrame) -> pd.DataFrame:
    """Eagerly materialised whenever raw_orders updates."""
    ...


@asset(
    partitions_def=daily,
    automation_condition=AutomationCondition.eager(),
    io_manager_key="snowflake_io",
    group_name="mart",
    compute_kind="snowflake",
)
def daily_revenue(context, orders_clean: pd.DataFrame) -> pd.DataFrame:
    """Eagerly materialised whenever orders_clean updates."""
    ...


@asset(
    partitions_def=daily,
    automation_condition=AutomationCondition.eager() & AutomationCondition.all_deps_updated(),
    io_manager_key="snowflake_io",
    group_name="mart",
    compute_kind="snowflake",
)
def weekly_revenue(context, daily_revenue: pd.DataFrame) -> pd.DataFrame:
    """Eagerly *and* only when all 7 upstream daily partitions are fresh."""
    ...

Step-by-step explanation.

1. raw_orders has automation_condition=AutomationCondition.on_cron("0 * * * *"). The runtime's automation evaluator runs every ~30 seconds; when the wall-clock crosses a cron tick, the condition is true and a materialization is launched for the current partition.
2. orders_clean has AutomationCondition.eager(). When raw_orders completes, the runtime notes the upstream materialization event; on the next evaluator tick, eager() resolves true on the corresponding downstream partition, and a materialization is launched. Latency between root and clean: 30 seconds worst-case.
3. daily_revenue also has AutomationCondition.eager(). Same behaviour cascading from orders_clean. The chain is self-driving; no @schedule for the downstream.
4. weekly_revenue uses eager() & all_deps_updated(). eager() alone would fire whenever any upstream day is updated; combining with all_deps_updated() waits until all seven upstream daily partitions in the week are fresh. This is how you express "roll up weekly only when the week is complete."
5. Over a wall-clock hour, the runtime launches exactly one raw_orders run at :00, cascading into one orders_clean and one daily_revenue. weekly_revenue only fires once per week when the 7th day completes. Total runs per hour = 3 (or 4 on the weekly tick), all driven by conditions, zero scheduled explicitly.

Output.

Wall clock	Event	Runs launched by evaluator
14:00:00	cron tick on raw_orders	raw_orders[2026-07-15]
14:03:00	raw_orders done	orders_clean[2026-07-15]
14:06:00	orders_clean done	daily_revenue[2026-07-15]
14:08:00	daily_revenue done	weekly_revenue only if all 7 days fresh
15:00:00	next cron tick	raw_orders[2026-07-15] again

Rule of thumb. For any pipeline whose downstream should react to upstream freshness, use AutomationCondition.eager() on downstream assets and let one on_cron on the root drive the cadence. Do not layer redundant @schedule blocks on every asset — that fights the runtime.

Worked example — composed conditions with rate limiting and back-pressure

Detailed explanation. A senior deployment often needs more nuance than plain eager(). Consider: expensive_asset reads a 100 GB upstream, should react to upstream but at most every 30 minutes to control cost, must never run in parallel with itself, and should skip if the previous run failed (until an operator explicitly retries). Walk through the composed AutomationCondition.

• Upstream reactivity. any_deps_updated().
• Rate limit. since_last_handled(minutes=30) — at most every 30 minutes.
• Serial only. ~in_progress() — never overlap.
• Skip on prior fail. manual retry via UI (implicit; condition just waits).

Question. Compose the condition and show how the runtime evaluates it.

Input.

Requirement	Atom
React to upstream	any_deps_updated()
Rate limit 30 min	since_last_handled(minutes=30)
Never overlap	~in_progress()
Backfill new partitions	on_missing()

Code.

from dagster import asset, AutomationCondition, DailyPartitionsDefinition
from datetime import timedelta
import pandas as pd

daily = DailyPartitionsDefinition(start_date="2026-01-01")


def expensive_condition() -> AutomationCondition:
    """Compose: eager-ish but rate-limited to 30 min and never overlapping."""
    react = AutomationCondition.any_deps_updated().since_last_handled()
    fresh = AutomationCondition.on_missing()
    no_overlap = ~AutomationCondition.in_progress()

    # rate-limit: only allow another launch if it's been > 30 min since last request
    # Dagster expresses this via since_last_handled() combined with a cron pace.
    return (fresh | react) & no_overlap


@asset(
    partitions_def=daily,
    automation_condition=expensive_condition(),
    io_manager_key="snowflake_io",
    group_name="mart",
    compute_kind="snowflake",
    tags={"cost_tier": "expensive"},
)
def expensive_asset(context, orders_clean: pd.DataFrame) -> pd.DataFrame:
    """100GB Snowflake MERGE; cap concurrency + pace to 30 min."""
    ...

# dagster.yaml — pair with tag-based concurrency limits
run_coordinator:
  module: dagster.core.run_coordinator
  class:  QueuedRunCoordinator
  config:
    max_concurrent_runs: 20
    tag_concurrency_limits:
      - key: cost_tier
        value: expensive
        limit: 1                    # never more than 1 expensive_asset run
      - key: compute_kind
        value: snowflake
        limit: 5                    # at most 5 warehouse runs total

# automation_sensor_options.py — override the evaluator tick and set the pace
from dagster import Definitions, AutomationConditionSensorDefinition, AssetSelection

automation_sensor = AutomationConditionSensorDefinition(
    name="expensive_asset_automation",
    asset_selection=AssetSelection.assets("expensive_asset"),
    minimum_interval_seconds=1800,       # 30 min pace on the evaluator itself
)

Step-by-step explanation.

1. The composed condition (on_missing() | any_deps_updated().since_last_handled()) & ~in_progress() reads as: "fire when (the partition is missing) or (an upstream has updated since we last requested a run), AND we aren't already running." Every atom is a first-class condition object; the & | ~ operators produce a new composite.
2. since_last_handled() is the modifier that prevents re-emission spam. Without it, any_deps_updated() would fire on every evaluator tick as long as upstream is fresher than downstream; with it, the condition scopes to "since we last emitted a request for this (asset, partition)."
3. ~AutomationCondition.in_progress() prevents launching a second run while the first is still in flight. This is critical for expensive assets — otherwise a stuck run + a fresh upstream update = two concurrent 100 GB Snowflake MERGEs.
4. tag_concurrency_limits in dagster.yaml caps cost_tier=expensive at 1 concurrent run globally. Even if expensive_condition() somehow fired twice, the run coordinator would queue the second. This is the belt-and-braces defense.
5. AutomationConditionSensorDefinition(..., minimum_interval_seconds=1800) slows the evaluator for this asset to a 30-minute tick. The default is ~30 seconds; overriding on cost-sensitive assets prevents unnecessary evaluation and matches the effective pace of the compute.

Output.

Scenario	Composed condition truth	Runtime action
Partition missing, no run in progress	true	launch
Upstream fresh, since_last_handled < 30 min	false (since_last_handled scopes)	wait
Upstream fresh, run in progress	false (~in_progress)	wait
Upstream fresh, since_last_handled > 30 min, ~in_progress	true	launch
Concurrent second launch would fire	true, but queue limit=1	queued

Rule of thumb. For expensive assets, always compose since_last_handled() (to prevent re-emission spam), ~in_progress() (to prevent overlap), and pair with tag_concurrency_limits in dagster.yaml (belt-and-braces). One-line eager() is fine for cheap assets; expensive ones deserve the four-line composite.

Worked example — sensor + schedule + automation together

Detailed explanation. A production Dagster deployment usually mixes three scheduling mechanisms: sensors for external events (a new S3 file), schedules for cron-driven job runs (a nightly full refresh), and Declarative Automation for the asset-graph baseline. Walk through a coherent example that uses all three without stepping on each other.

• Sensor. s3_new_file_sensor — reacts to new files in s3://acme/inbox/orders/.
• Schedule. nightly_full_refresh_schedule — 3 AM UTC, materialises a full_refresh_snapshot asset once per night.
• Automation. AutomationCondition.eager() on downstream mart assets.

Question. Define one of each, wire them into Definitions, and describe when each fires.

Input.

Mechanism	Trigger
@sensor s3_new_file_sensor	new key in s3://acme/inbox/orders/
@schedule nightly_full_refresh_schedule	0 3 * * *
AutomationCondition.eager()	any upstream mart update

Code.

# sensors.py — react to a new S3 file
from dagster import sensor, RunRequest, SkipReason, SensorEvaluationContext
import boto3

@sensor(minimum_interval_seconds=60)
def s3_new_file_sensor(context: SensorEvaluationContext):
    """Emit a RunRequest for the ingest job when a new key appears."""
    last_seen = context.cursor or ""
    s3 = boto3.client("s3")
    resp = s3.list_objects_v2(Bucket="acme", Prefix="inbox/orders/")
    keys = sorted(o["Key"] for o in resp.get("Contents", []))
    new_keys = [k for k in keys if k > last_seen]

    if not new_keys:
        yield SkipReason("no new files")
        return

    for k in new_keys:
        yield RunRequest(
            run_key=k,                       # dedupe by S3 key
            run_config={"ops": {"ingest_s3_file": {"config": {"key": k}}}},
        )

    context.update_cursor(new_keys[-1])

# schedules.py — nightly full refresh
from dagster import schedule, RunRequest, ScheduleEvaluationContext
from .jobs import nightly_full_refresh_job

@schedule(
    job=nightly_full_refresh_job,
    cron_schedule="0 3 * * *",
)
def nightly_full_refresh_schedule(context: ScheduleEvaluationContext):
    yield RunRequest(
        run_key=f"nightly-{context.scheduled_execution_time.date()}",
        tags={"trigger": "schedule", "kind": "full_refresh"},
    )

# definitions.py — everything registered
from dagster import Definitions
from .assets import raw_orders, orders_clean, daily_revenue, full_refresh_snapshot
from .sensors import s3_new_file_sensor
from .schedules import nightly_full_refresh_schedule
from .jobs import nightly_full_refresh_job, ingest_s3_file_job

defs = Definitions(
    assets=[raw_orders, orders_clean, daily_revenue, full_refresh_snapshot],
    sensors=[s3_new_file_sensor],
    schedules=[nightly_full_refresh_schedule],
    jobs=[nightly_full_refresh_job, ingest_s3_file_job],
    resources={...},
)

Step-by-step explanation.

1. s3_new_file_sensor evaluates every 60 seconds (minimum_interval_seconds=60). It lists S3, filters to keys strictly greater than the sensor's cursor, emits one RunRequest per new key (with run_key=k for dedup), and updates the cursor to the newest key. Duplicate emissions for the same key are silently deduped by the runtime.
2. nightly_full_refresh_schedule fires at 03:00 UTC via cron_schedule="0 3 * * *". It emits a RunRequest for nightly_full_refresh_job, which materialises the full_refresh_snapshot asset. Once per night, exactly.
3. The AutomationCondition.eager() on daily_revenue (and other downstream mart assets) continues to react to upstream materializations — whether those upstreams were driven by the sensor, the schedule, or another automation condition. All three mechanisms feed the same asset event log; the automation evaluator sees them uniformly.
4. The three mechanisms don't fight because they target different concerns: sensor = external event, schedule = time-driven job, automation = asset-graph freshness. Overlap would happen if you also put on_cron("0 3 * * *") on full_refresh_snapshot; the discipline is one mechanism per firing reason.
5. Every emission carries tags that show up in the UI's run list: trigger=sensor, trigger=schedule, or (implicit) trigger=automation. The on-call filters by tag when debugging "who launched this run?"

Output.

Firing mechanism	Cadence	Emission
s3_new_file_sensor	every 60s if new keys	1 RunRequest per new key
nightly_full_refresh_schedule	once at 03:00 UTC	1 RunRequest per night
AutomationCondition.eager() on daily_revenue	evaluator tick	1 run per upstream materialization

Rule of thumb. Use sensors for genuinely external events, schedules for job-level cron runs, and Declarative Automation for the asset-graph baseline. One mechanism per firing reason; never let two mechanisms target the same materialization or you'll get spurious double-launches.

Senior interview question on Declarative Automation

A senior interviewer might ask: "You have a 40-asset analytics graph with one hourly root, twenty daily downstream marts, and one weekly rollup. Design the automation strategy — which assets carry on_cron, which are eager, which need all_deps_updated, and how do you prevent the weekly rollup from firing mid-week? Also cover the run-coordinator config for cost control."

Solution Using one on_cron root, an eager cascade, a strict weekly gate, and tag-based rate limiting

# automation.py — the coherent 40-asset automation strategy
from dagster import (
    asset,
    AutomationCondition,
    DailyPartitionsDefinition,
    WeeklyPartitionsDefinition,
    AssetIn,
    TimeWindowPartitionMapping,
)
import pandas as pd

daily  = DailyPartitionsDefinition(start_date="2026-01-01")
weekly = WeeklyPartitionsDefinition(start_date="2026-01-05")   # Monday start


# Root: hourly cron
@asset(
    partitions_def=daily,
    automation_condition=AutomationCondition.on_cron("0 * * * *"),
    io_manager_key="s3_parquet_io",
    group_name="raw",
    tags={"tier": "root"},
)
def raw_events(context) -> pd.DataFrame: ...


# 20 downstream daily marts — eager with rate-limit + no overlap
def mart_condition() -> AutomationCondition:
    return AutomationCondition.eager() & ~AutomationCondition.in_progress()

def mart_asset(name: str):
    @asset(
        name=name,
        partitions_def=daily,
        automation_condition=mart_condition(),
        io_manager_key="snowflake_io",
        group_name="mart",
        tags={"tier": "mart", "compute": "snowflake"},
    )
    def _fn(context, raw_events: pd.DataFrame) -> pd.DataFrame: ...
    return _fn

mart_assets = [mart_asset(f"mart_{i:02d}") for i in range(1, 21)]


# Weekly rollup — strict all_deps_updated over 7 days
@asset(
    partitions_def=weekly,
    ins={
        "mart_01": AssetIn(partition_mapping=TimeWindowPartitionMapping()),
    },
    automation_condition=(
        AutomationCondition.eager()
        & AutomationCondition.all_deps_updated()
        & ~AutomationCondition.in_progress()
    ),
    io_manager_key="snowflake_io",
    group_name="rollup",
    tags={"tier": "weekly", "compute": "snowflake"},
)
def weekly_rollup(context, mart_01: dict[str, pd.DataFrame]) -> pd.DataFrame:
    """Fires only when all 7 daily partitions of mart_01 are fresh."""
    return pd.concat(mart_01.values(), ignore_index=True)

# dagster.yaml — cost-control run coordinator
run_coordinator:
  module: dagster.core.run_coordinator
  class:  QueuedRunCoordinator
  config:
    max_concurrent_runs: 30
    tag_concurrency_limits:
      - key: compute
        value: snowflake
        limit: 8              # at most 8 parallel Snowflake runs
      - key: tier
        value: weekly
        limit: 1              # only one weekly rollup at a time
      - key: tier
        value: mart
        limit: 12             # cap mart-tier parallelism

Step-by-step trace.

Layer	Condition	Behaviour
Root (raw_events)	on_cron("0 * * * *")	fire every hour
20 marts	eager() & ~in_progress()	cascade after root; no overlap
Weekly rollup	eager() & all_deps_updated() & ~in_progress()	only when all 7 days fresh
Coordinator cap: snowflake	8	at most 8 warehouse runs
Coordinator cap: weekly	1	serial rollup
Coordinator cap: mart	12	bounded fan-out

After deployment, one root on_cron drives 20 downstream marts eagerly; the marts are cost-capped at 12 parallel. The weekly rollup waits until Sunday when all 7 upstream days are fresh, then fires exactly once. Every run's tag limits are enforced by the coordinator; no automation condition can flood the warehouse.

Output:

Metric	Value
Explicit schedules	1 (the root on_cron)
Automation conditions	22 (20 marts + weekly + root)
Mart parallelism cap	12
Snowflake parallelism cap	8
Weekly cadence	1x/week when all deps fresh
Runs per hour, steady state	1 root + up to 12 marts

Why this works — concept by concept:

• One on_cron root — the entire cadence of the graph derives from one cron. Downstream assets don't need their own schedules; they react. This is the cleanest separation of "when the data appears" from "when the derived data appears."
• eager() & ~in_progress() — the standard mart-tier condition. Reacts to upstream, forbids overlap. ~in_progress() is the cheap guardrail against runaway automation when a stuck run blocks a fresh upstream.
• all_deps_updated() for weekly — the strict gate. Without it, the weekly rollup would fire every time any daily partition updates — 7 times a week instead of 1. all_deps_updated() is the primitive for "wait for the whole window."
• tag_concurrency_limits — the cost cap. compute=snowflake limit 8 ensures the entire Dagster deployment cannot spin up more than 8 Snowflake warehouses in parallel. tier=weekly limit 1 ensures rollups are serial. tier=mart limit 12 caps fan-out.
• Cost — 22 typed conditions (5-10 lines each), one dagster.yaml block with three tag limits, one Postgres for the Dagster instance. The eliminated cost is the per-team schedule and the ad-hoc "we forgot to add a semaphore" scaffolding that most Airflow shops accrete. O(assets) per evaluator tick; O(1) per condition evaluation. Compared to writing 22 sensors + 22 schedules by hand, this is a factor-of-10 reduction in orchestration surface area.

Design
Topic — design
Design problems on declarative automation

Practice →

Streaming Topic — streaming Streaming problems for sensor-driven pipelines

Practice →

---

5. Dagster vs Airflow migration and interview signals

dagster vs airflow is asset-centric vs task-centric — and migration is a per-team, dagster-airflow-bridged, 12-to-18-month project

The mental model in one line: dagster vs airflow is not a "which scheduler wins" question — it's a "which mental model does the platform describe itself in" question, and the migration path from Airflow to Dagster is per-team, incremental, and bridged by the dagster-airflow package that lets Dagster assets shell out to Airflow DAGs (or Airflow tasks trigger Dagster assets) while both systems run side by side; senior interviewers score highest on answers that concede Airflow's ecosystem win, name the specific bridge library, sketch the team-by-team cutover, and acknowledge that migration is a 12-18-month project rather than a weekend rewrite. The dagster-dbt deep integration is the single largest wedge because dbt models are the natural first tier to migrate.

The four-axis Dagster-vs-Airflow decision matrix.

• Unit of work. Airflow: Task / Operator. Dagster: @asset (SDA). This is the load-bearing distinction; every other axis flows from it.
• Data lineage. Airflow: none built-in; requires OpenLineage or a separate catalog (DataHub, Amundsen). Dagster: native asset graph; every materialization event carries lineage.
• Data quality integration. Airflow: GreatExpectationsOperator as a separate task. Dagster: @asset_check inside the same event; blocking or non-blocking policies.
• dbt integration. Airflow: DbtOperator runs dbt as a shell step; you don't see individual models in the UI. Dagster: @dbt_assets(manifest=...) produces one Dagster asset per dbt model with full lineage.

When Airflow still wins in 2026.

• Operator ecosystem breadth. 500+ community + provider operators. If your workload is 90% "call this operator on this cron," Airflow gets you shipping faster.
• Existing platform mass. A 400-DAG Airflow deployment cannot rewrite itself; the migration cost is real. Airflow keeps winning by inertia in shops where the existing tooling is deeply invested.
• Cross-vendor operator library. Snowflake, BigQuery, Databricks, Salesforce, Snowplow, etc. all have first-class Airflow operators. Dagster covers the top-tier vendors well; the long tail is thinner.
• Battle-tested at 10,000-DAG scale. Airbnb, Lyft, Airbnb-scale deployments have been running Airflow for a decade. Dagster is production-hardened but the largest known deployments are in the low thousands of assets, not tens of thousands.

The three migration paths.

• dagster-airflow bridge. Dagster asset shells out to an Airflow DAG (or Airflow task triggers a Dagster asset). Lets you run both platforms in parallel, one team at a time. Slowest but safest; the default recommendation.
• dagster-embedded-elt for EL/T sources. Migrate ingestion (Fivetran/Sling/Airbyte replacements) first via dagster-embedded-elt, then work upward through the asset graph. Wedge play.
• Big-bang rewrite. Only viable for small deployments (< 50 DAGs). Anyone attempting big-bang at scale ends up with two half-migrated platforms and technical debt on both sides.

The dagster-dbt deep integration — the migration wedge.

• What it does. Parses dbt_project/target/manifest.json at code-load time and creates one Dagster asset per dbt model. Every model's config, tags, and lineage flow into the Dagster UI.
• The equivalent in Airflow. DbtOperator (or Astronomer Cosmos) runs dbt build as a task; you don't see individual models. Cosmos improved this by parsing the manifest and creating one Airflow task per model, but the UI is still task-graph-first.
• Why it matters. dbt is the fastest-growing analytics-engineering tool in 2026. Any team with a substantial dbt project gets more mileage from Dagster's model-as-asset UI than from Airflow's task-per-model UI. This is the wedge — migrate dbt to Dagster first.
• How. @dbt_assets(manifest=...) — one decorator wraps the whole dbt project. Add select=... to scope which models load into Dagster. Add partitions_def=... to attach a partition scheme (Dagster passes the partition key as a dbt variable).

Common interview probes on migration.

• "When would you keep Airflow?" — required answer: existing large deployment + operator-heavy workload + no dbt-first tier.
• "How would you migrate from Airflow to Dagster?" — required answer: dagster-airflow bridge; per-team; start with dbt tier.
• "How does dagster-dbt differ from DbtOperator?" — deep parse of manifest.json vs shell out to dbt build; asset-per-model vs task-per-DAG.
• "How long would a 400-DAG migration take?" — required answer: 12-18 months, one team at a time.

Worked example — mapping an Airflow DAG to a Dagster asset graph

Detailed explanation. Take a canonical Airflow DAG — extract from RDS, transform with pandas, load to Snowflake, publish a Slack success message. Show the direct 1:1 Dagster asset graph rewrite, then discuss what actually improves. Walk through both.

• Airflow DAG. 4 tasks: extract, transform, load, notify.
• Dagster rewrite. 3 assets (extract, transform, load) + 1 sensor for notify.
• What improves. Native lineage, asset checks, partition-per-day out of the box.

Question. Rewrite the 4-task Airflow DAG as a Dagster asset graph, and enumerate the observability improvements.

Input.

Airflow	Dagster
extract_rds task	raw_from_rds asset
transform_pandas task	transformed_df asset
load_snowflake task	snowflake_orders asset
notify_slack task	@run_status_sensor on job success
DailyOperator schedule	DailyPartitionsDefinition + on_cron
Task lineage	Native asset graph
Data quality	AssetCheckSpec blocking

Code.

# airflow_original.py — the DAG being migrated
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.slack.operators.slack import SlackAPIPostOperator
from datetime import datetime

def extract(**_):    ...
def transform(**_):  ...
def load(**_):       ...

with DAG("orders_pipeline", start_date=datetime(2026, 1, 1),
         schedule="0 3 * * *", catchup=False) as dag:
    t1 = PythonOperator(task_id="extract",   python_callable=extract)
    t2 = PythonOperator(task_id="transform", python_callable=transform)
    t3 = PythonOperator(task_id="load",      python_callable=load)
    t4 = SlackAPIPostOperator(task_id="notify", text="orders_pipeline done")
    t1 >> t2 >> t3 >> t4

# dagster_rewrite.py — same behaviour as Dagster assets
from dagster import (
    asset,
    asset_check,
    AssetCheckResult,
    AssetCheckSpec,
    DailyPartitionsDefinition,
    AutomationCondition,
    run_status_sensor,
    DagsterRunStatus,
    RunRequest,
    Definitions,
)
import pandas as pd

daily = DailyPartitionsDefinition(start_date="2026-01-01")


@asset(
    partitions_def=daily,
    automation_condition=AutomationCondition.on_cron("0 3 * * *"),
    io_manager_key="s3_parquet_io",
    group_name="raw",
    compute_kind="rds",
)
def raw_from_rds(context) -> pd.DataFrame:
    """Extract from RDS; runs at 03:00 UTC via cron."""
    ...


@asset(
    partitions_def=daily,
    automation_condition=AutomationCondition.eager(),
    io_manager_key="s3_parquet_io",
    group_name="transform",
    compute_kind="pandas",
)
def transformed_df(context, raw_from_rds: pd.DataFrame) -> pd.DataFrame:
    """Pandas transform; eager after RDS extract."""
    ...


@asset(
    partitions_def=daily,
    automation_condition=AutomationCondition.eager(),
    io_manager_key="snowflake_io",
    group_name="load",
    compute_kind="snowflake",
    check_specs=[
        AssetCheckSpec(name="row_count_gt_zero", asset="snowflake_orders", blocking=True),
    ],
)
def snowflake_orders(context, transformed_df: pd.DataFrame) -> pd.DataFrame:
    """Load to Snowflake with a blocking row-count check."""
    ...


# Notify slack when the *asset* materialisation succeeds
@run_status_sensor(
    run_status=DagsterRunStatus.SUCCESS,
    monitored_jobs=None,  # monitor everything, filter in the handler
)
def notify_slack_on_success(context):
    if "snowflake_orders" not in (context.dagster_run.asset_selection or set()):
        return
    # send slack message
    ...


defs = Definitions(
    assets=[raw_from_rds, transformed_df, snowflake_orders],
    sensors=[notify_slack_on_success],
    resources={...},
)

Step-by-step explanation.

1. The four Airflow tasks become three Dagster assets plus one run-status sensor. The extract → transform → load chain is expressed as dependencies via function parameters; no >> operator needed. The notify_slack task becomes a @run_status_sensor that fires after the materialization succeeds — decoupling notification from the pipeline.
2. The Airflow schedule "0 3 * * *" becomes AutomationCondition.on_cron("0 3 * * *") on the root asset. The two downstream assets use AutomationCondition.eager() to react automatically; you don't schedule them explicitly. That's a big observability improvement — the runtime "knows" which downstream needs to run.
3. Airflow's catchup=False semantics are preserved because Dagster doesn't backfill without an explicit backfill action. If you want the catchup behaviour, you'd add AutomationCondition.on_missing() to fill missing partitions automatically.
4. The blocking AssetCheckSpec(name="row_count_gt_zero", asset="snowflake_orders", blocking=True) gives you a data-quality gate the Airflow version doesn't have. In the Airflow world, adding this would mean a separate GreatExpectationsOperator task; in Dagster, it's one line in the asset definition.
5. The observability upgrade is the biggest win. Airflow shows you a 4-node task graph with green/red dots per task run. Dagster shows you a 3-node asset graph with green/red dots per materialization plus row counts, freshness, check status, and native lineage into downstream assets that this DAG feeds.

Output.

Feature	Airflow	Dagster
Nodes	4 tasks	3 assets + 1 sensor
Schedule	schedule="0 3 * * *"	on_cron("0 3 * * *") + eager cascade
Lineage	none (needs OpenLineage)	native asset graph
Row count observability	via XCom + Grafana	inline MetadataValue.int
Data quality gate	separate GE task	AssetCheckSpec inline
Backfill	airflow dags backfill per DAG	UI action per asset
Cross-DAG dependency	ExternalTaskSensor	asset graph edge

Rule of thumb. When rewriting an Airflow DAG as Dagster assets, count the nodes: N tasks usually becomes N-1 assets (the "load" and "extract" collapse into ends) plus one sensor for out-of-band side effects (notify). Fewer nodes, more observability, more automatic dependency management.

Worked example — dagster-airflow bridge for a parallel-run migration

Detailed explanation. For a 400-DAG Airflow shop, the migration is per-team, and the two systems need to coexist for a year plus. dagster-airflow provides two bridge modes: (a) a Dagster asset that shells out to an Airflow DAG (Dagster owns the schedule; Airflow does the compute), and (b) an Airflow operator that triggers a Dagster asset materialization (Airflow owns the schedule; Dagster does the compute). Walk through both.

• Direction 1. Dagster asset triggers Airflow DAG.
• Direction 2. Airflow operator triggers Dagster asset.
• Rollout plan. Team-by-team; each team picks a direction based on which system owns their scheduling.

Question. Wire both directions with dagster-airflow and describe the per-team cutover.

Input.

Direction	Use case
Dagster → Airflow DAG	teams that want asset lineage + Airflow DAG still runs the compute
Airflow → Dagster asset	teams whose schedule + operator library stays; wants asset observability

Code.

# direction 1: dagster asset that runs an airflow dag
from dagster import asset, DailyPartitionsDefinition
from dagster_airflow import load_assets_from_airflow_dag, make_dagster_definitions_from_airflow_dag_bag

daily = DailyPartitionsDefinition(start_date="2026-01-01")

# Option A: point at an existing DAG file and let dagster-airflow generate assets
airflow_dag_defs = make_dagster_definitions_from_airflow_dag_bag(
    dag_bag_path="/opt/airflow/dags",
    tags={"origin": "airflow_migrated"},
)

# Option B: shell out to a running Airflow instance via REST
import requests

@asset(
    partitions_def=daily,
    automation_condition=None,   # this asset is only triggered by other Dagster events
    group_name="airflow_bridged",
)
def trigger_orders_pipeline_airflow_dag(context) -> None:
    """Materialise this asset → trigger the legacy Airflow DAG."""
    day = context.partition_key
    r = requests.post(
        "http://airflow:8080/api/v1/dags/orders_pipeline/dagRuns",
        json={"conf": {"execution_date": day}, "logical_date": f"{day}T03:00:00Z"},
        auth=("dagster", "..."),
        timeout=30,
    )
    r.raise_for_status()
    context.log.info(f"triggered airflow DAG for {day}")

# direction 2: airflow operator that materialises a dagster asset
from airflow.providers.http.operators.http import SimpleHttpOperator

trigger_dagster = SimpleHttpOperator(
    task_id="materialise_snowflake_orders",
    http_conn_id="dagster",
    endpoint="graphql",
    method="POST",
    data='''{
      "query": "mutation { launchPipelineExecution(executionParams: { selector: { repositoryLocationName: \\"analytics\\", repositoryName: \\"__repository__\\", pipelineName: \\"asset_group\\" }, mode: \\"default\\" }) { __typename } }"
    }''',
)

# Per-team cutover plan (12-18 months for 400 DAGs / 20 teams)
Month 1-2   — set up Dagster + `dagster-airflow`, one Definitions
Month 3-4   — team A (analytics/dbt) fully on Dagster; Airflow reads Dagster
Month 5-6   — team B (ML feature engineering) on Dagster
Month 7-8   — team C, D
Month 9-12  — remaining teams migrate; Airflow still runs long-tail DAGs
Month 12-18 — final teams migrate; Airflow decommissioned or kept for external triggers

Step-by-step explanation.

1. make_dagster_definitions_from_airflow_dag_bag scans an Airflow DAG folder and generates one Dagster asset per DAG. The asset's materialization function calls the Airflow REST API to trigger a DAG run. This is the "Dagster owns the schedule, Airflow does the compute" pattern.
2. Alternatively, an @asset can shell out via requests.post to Airflow's REST API directly (Option B). This gives you more control over which DAGs to expose (only the ones you want migrated first) and how to translate partition keys into Airflow logical_date.
3. For the other direction, an Airflow SimpleHttpOperator (or a Dagster-Airflow-provided operator) POSTs a GraphQL mutation to Dagster's GraphQL API to launch a materialization. This is the "Airflow owns the schedule, Dagster does the compute" pattern — useful when the team's cron logic lives in Airflow but the compute belongs in a Dagster asset.
4. The per-team cutover works because both systems can point at the same source data. Team A migrates dbt to dagster-dbt while Team B's pandas pipeline stays in Airflow. Neither team is blocked on the other; both systems see the same warehouse.
5. Sequencing tips: (a) start with the highest-leverage team (usually analytics/dbt because dagster-dbt is such a clean fit); (b) migrate teams whose upstreams are already in Dagster next (fewer bridge calls); (c) leave the operator-heavy long-tail (Salesforce, Snowplow, etc.) on Airflow until the end because those operators don't have Dagster equivalents.

Output.

Phase	Airflow role	Dagster role
Month 0	400 DAGs, everything	none
Month 3	380 DAGs + bridge	dbt tier (20 DAGs → 200 assets)
Month 8	200 DAGs + bridge	analytics + ML + streaming teams
Month 12	50 DAGs + bridge	350+ assets
Month 18	0 DAGs (decommission) or specialised triggers only	full analytics platform

Rule of thumb. For any Airflow-to-Dagster migration over 100 DAGs, use dagster-airflow bridges, migrate one team at a time, sequence dbt-heavy teams first, and expect 12-18 months of parallel operation. A weekend rewrite for 400 DAGs is not a plan; it's a resume-generating event.

Worked example — dagster-dbt deep integration as the migration wedge

Detailed explanation. The single fastest way to demonstrate Dagster's value in an Airflow shop is to migrate the dbt tier. dagster-dbt parses manifest.json and creates one Dagster asset per dbt model with correct lineage, tags, and materialization events — a step-change improvement over Airflow's DbtOperator (which runs dbt build as one task) or even Cosmos (which creates one Airflow task per model but keeps a task-graph UI).

• Source. A dbt project with ~200 models across staging, intermediate, marts.
• Wire-in. One @dbt_assets decorator + one Definitions.
• Result. 200 Dagster assets, full lineage, per-model materialization events.

Question. Wire dagster-dbt into an existing dbt project and describe what appears in the Dagster UI.

Input.

Component	Value
dbt project	analytics_dbt/
Manifest path	analytics_dbt/target/manifest.json
Models	~200 across staging/intermediate/marts
Partition scheme	DailyPartitionsDefinition for date-partitioned marts

Code.

# assets.py — the whole dbt project as Dagster assets
from dagster import (
    AssetExecutionContext,
    DailyPartitionsDefinition,
    Definitions,
)
from dagster_dbt import DbtCliResource, dbt_assets, DagsterDbtTranslator
import os

DBT_PROJECT_DIR = os.path.expanduser("~/analytics_dbt")
DBT_MANIFEST = f"{DBT_PROJECT_DIR}/target/manifest.json"

daily = DailyPartitionsDefinition(start_date="2026-01-01")


class AnalyticsDbtTranslator(DagsterDbtTranslator):
    """Give date-partitioned mart models a partition_def; others none."""
    def get_partition_mapping(self, dbt_resource_props):
        tags = dbt_resource_props.get("tags", [])
        if "partitioned_daily" in tags:
            return daily
        return None

    def get_group_name(self, dbt_resource_props):
        return dbt_resource_props["fqn"][-2]   # dbt subdirectory as group

    def get_tags(self, dbt_resource_props):
        return {"origin": "dbt", "materialization": dbt_resource_props["config"]["materialized"]}


@dbt_assets(
    manifest=DBT_MANIFEST,
    dagster_dbt_translator=AnalyticsDbtTranslator(),
)
def analytics_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()


defs = Definitions(
    assets=[analytics_dbt_assets],
    resources={
        "dbt": DbtCliResource(project_dir=DBT_PROJECT_DIR, profiles_dir=DBT_PROJECT_DIR),
    },
)

# CI hook — regenerate manifest.json on every dbt project change
cd analytics_dbt
dbt parse                             # produces target/manifest.json
git add target/manifest.json
git commit -m "regen manifest for dagster-dbt asset load"

# On the Dagster side, reloading the code location re-parses manifest.json
dagster asset materialize --select 'staging_orders'

Step-by-step explanation.

1. @dbt_assets(manifest=DBT_MANIFEST) parses the dbt manifest at Dagster code-load time. Every model in the manifest becomes one Dagster asset, keyed by the model's fully-qualified dbt name (staging.stg_orders, intermediate.int_customer_orders, marts.fct_orders).
2. The DagsterDbtTranslator subclass customises the mapping. get_partition_mapping attaches DailyPartitionsDefinition only to models tagged partitioned_daily in dbt (via {{ config(tags=['partitioned_daily']) }}). get_group_name uses the dbt subdirectory as the Dagster group. get_tags propagates dbt materialization type (view, table, incremental) into Dagster tags.
3. Inside analytics_dbt_assets, dbt.cli(["build"], context=context).stream() runs dbt build and streams per-model events. Each dbt model's start / success / failure translates into a Dagster materialization event, so the UI shows per-model status in real time — not just a single "did dbt succeed" bit.
4. The dbt lineage propagates into Dagster: if fct_orders depends on stg_orders in dbt's manifest, that edge appears in Dagster's asset graph automatically. Adding a non-dbt Dagster asset upstream of stg_orders (via deps=[stg_orders]) makes the dbt graph extend into the non-dbt world seamlessly.
5. The CI hook regenerates manifest.json on every dbt change so the Dagster code location picks up new/changed models on next reload. In Dagster+ this happens automatically per deployment; in OSS Dagster, reload the code location manually or wire into your deployment pipeline.

Output.

Dagster UI element	With dagster-dbt
Asset count	~200 (one per dbt model)
Lineage	native (parsed from manifest)
Per-model materialization events	yes
Per-model tags	dbt tags + materialization type
Grouping	by dbt subdirectory (staging/int/marts)
Partitions	attached per model via translator rules

Rule of thumb. For any Airflow shop with a significant dbt tier, prioritise the dagster-dbt migration first — the step-change in per-model observability is the single largest demonstrable value of Dagster, and the migration is one decorator plus a translator. Everything downstream of the dbt tier can migrate at leisure.

Senior interview question on migration

A senior interviewer might ask: "You are the tech lead of a 30-engineer platform team running 400 Airflow DAGs across 20 sub-teams — including a substantial dbt tier (200 models), an ML feature engineering group, and a Kafka streaming team. Design a 12-18-month migration to Dagster: wedge strategy, dagster-airflow bridge, per-team sequencing, rollback plan, and what you'd tell the CTO about ecosystem trade-offs."

Solution Using a dbt-first wedge, dagster-airflow bidirectional bridge, and a per-team quarterly cutover plan

# migration_plan.py — the phased Dagster rollout for 400 DAGs
"""
Migration axes:
  - wedge:         dbt tier first (dagster-dbt deep integration)
  - bridge:        dagster-airflow bidirectional
  - sequencing:    team-by-team, quarterly
  - rollback:      per-team; symlinked config in git; < 1 day
  - decommission:  Airflow stays as long as one DAG remains, then hard-off
"""

MIGRATION_PLAN = [
    # Q1 — infra + dbt tier (wedge)
    {"quarter": "Q1", "team": "platform+analytics",
     "milestone": "Dagster+ deployed; `dagster-dbt` loads 200 dbt models"},

    # Q2-Q3 — analytics-adjacent teams
    {"quarter": "Q2", "team": "analytics_downstream",
     "milestone": "50 Airflow DAGs migrated to Dagster assets; bridge for legacy"},
    {"quarter": "Q3", "team": "ml_feature_eng",
     "milestone": "40 feature-store pipelines on Dagster assets + partitions"},

    # Q4-Q6 — streaming + long-tail
    {"quarter": "Q4", "team": "kafka_streaming",
     "milestone": "streaming ingests bridged via Dagster sensor + dagster-embedded-elt"},
    {"quarter": "Q5", "team": "long_tail_ops",
     "milestone": "80 operator-heavy DAGs stay on Airflow, bridged from Dagster asset graph"},
    {"quarter": "Q6", "team": "final_cleanup",
     "milestone": "Airflow decommissioned or retained for 20 external-trigger DAGs only"},
]

# dagster_cloud.yaml + per-team config in git
locations:
  - location_name: analytics
    code_source:
      package_name: acme_analytics
  - location_name: ml_features
    code_source:
      package_name: acme_ml
  - location_name: streaming
    code_source:
      package_name: acme_streaming

# Airflow bridge points (kept in the same repo)
# airflow_bridge.yaml
bridged_dags:
  - dag_id: legacy_salesforce_sync
    trigger_from: dagster_asset
    dagster_asset_key: [external, salesforce_daily_sync]
  - dag_id: analytics_backfill_v1
    trigger_from: airflow_dag
    dagster_target: refresh_snowflake_mart

# rollback_contract.py — per-team quick rollback
"""
Per-team rollback (< 1 day):
  1. Revert the PR that migrated the team's DAGs.
  2. Restore the Airflow DAGs to `scheduling=on` in the Airflow UI.
  3. Toggle the Dagster asset's automation_condition to None (freeze).
  4. Re-enable the Airflow schedule; disable Dagster's `on_cron` root.
  5. Monitor for 24h; if healthy, close the incident.
"""

Step-by-step trace.

Quarter	Team	Migration action	Airflow state after	Dagster state after
Q1	platform + analytics	deploy Dagster+, dagster-dbt	400 DAGs still on	200 dbt models as assets
Q2	analytics_downstream	50 DAGs migrated	350 DAGs	250 assets total
Q3	ml_feature_eng	40 pipelines migrated	310 DAGs	290 assets
Q4	kafka_streaming	bridge via sensor	290 DAGs	320+ assets
Q5	long_tail_ops	80 DAGs bridged (not migrated)	210 DAGs (bridged)	400+ assets
Q6	final cleanup	Airflow → 20 external triggers or decommission	20 or 0 DAGs	500 assets

After the 18-month rollout, the platform runs 500+ Dagster assets across three code locations; Airflow retains 20 external-trigger DAGs (or is decommissioned entirely). Every team migrated at their own pace via dagster-airflow bridges; rollback per team is a revert PR plus an Airflow schedule flip; the CTO gets one operational platform per team plus a bridged tail.

Output:

Metric	Value
Total migration duration	12-18 months
Per-team cutover cost	~1 quarter (analytics teams); ~2 quarters (long-tail)
Rollback SLA	< 1 day per team
Dagster asset count post-migration	500+
Airflow DAG count post-migration	0-20 (external triggers only)
Ecosystem concession	Airflow wins on 500+ operators; conceded up-front

Why this works — concept by concept:

• dbt-first wedge — dagster-dbt is the single largest demonstrable value; migrating the dbt tier in Q1 gives the platform team a big win and a natural upstream for downstream teams to target.
• dagster-airflow bidirectional bridge — both systems run in parallel; teams migrate at their own pace; long-tail operator-heavy DAGs never have to migrate at all if the bridge is stable enough.
• Per-team quarterly sequencing — no team is blocked on another; the platform team owns the bridge library; each team owns its cutover PR. Quarterly cadence matches team capacity for a non-trivial change.
• Rollback SLA per team — every migration is a git PR that can be reverted; the parallel Airflow scheduling means "revert and re-enable Airflow" is < 1 day per team. This is the risk-tolerance concession that lets the CTO approve the plan.
• Cost — one Dagster+ deployment, one bridge library (~500 lines), one repo of per-team configs, quarterly cutover coordination. The eliminated cost is the "big-bang migration failure" and the "half the team is on the old platform, half on the new" hybrid limbo that hurts every migration attempted without a bridge. O(teams) per quarter rather than O(DAGs) per weekend — for 20 teams that's a manageable per-quarter cost, not a Sisyphean rewrite.

Design
Topic — design
Design problems on orchestration migration

Practice →

SQL
Topic — sql
SQL practice for dbt-tier mart queries

Practice →

---

Cheat sheet — Dagster recipes

• Which orchestrator when. Dagster is the 2026 default for any greenfield analytics/ML/data-platform where the unit of work is naturally a dataset (dbt model, S3 partition, warehouse mart, feature table); its asset-centric mental model, dagster-dbt deep integration, and native lineage/quality primitives buy you observability that Airflow requires plugins for. Airflow is the answer when your workload is operator-heavy (500+ operators, integrating dozens of SaaS APIs), when you have an existing 200+-DAG platform that cannot pay the migration cost yet, or when your team's mental model is genuinely task-first. Prefect is the answer when you want "just Python" with the lightest footprint. Argo is the answer when the primary compute is a Kubernetes container and orchestration is a scheduling concern rather than a data-modelling one.
• @asset boilerplate. from dagster import asset, DailyPartitionsDefinition, MaterializeResult, MetadataValue; @asset(partitions_def=DailyPartitionsDefinition(start_date="2026-01-01"), io_manager_key="s3_parquet_io", group_name="raw", compute_kind="pandas", check_specs=[AssetCheckSpec(name="row_count_gt_zero", asset="my_asset", blocking=True)], tags={"team": "analytics"}) def my_asset(context, upstream_asset: pd.DataFrame) -> MaterializeResult: .... Return a MaterializeResult(value=..., metadata={...}, check_results=[...]) when you want inline metadata; return a raw value when the IO manager handles everything. Every knob is optional; the one-line minimal is @asset def x(): return 42.
• Partition definitions. DailyPartitionsDefinition(start_date="2026-01-01") for daily; HourlyPartitionsDefinition(start_date="2026-01-01-00:00") for hourly; StaticPartitionsDefinition(["us-east", "us-west"]) for fixed set; DynamicPartitionsDefinition(name="customer_id") for runtime-added; MultiPartitionsDefinition({"region": regions, "day": daily}) for region-x-day. Always give a start_date for time-window partitions — you almost never want an unbounded backfill window. Attach partitions on the asset via partitions_def=.
• Partition mappings. TimeWindowPartitionMapping() for hourly-into-daily fan-in; LastPartitionMapping() for "read the most recent upstream partition"; AllPartitionMapping() for "read every upstream partition" (careful — O(all history)); MultiToSingleDimensionPartitionMapping(partition_dimension_name="day") for multi-dim into single-dim fan-in. Declare explicitly via ins={"upstream": AssetIn(partition_mapping=...)} — the runtime infers reasonable defaults for identity mappings but always states multi-dim or non-identity mappings explicitly.
• IO manager S3 template. class S3ParquetIOManager(ConfigurableIOManager): bucket: str; prefix: str; def handle_output(self, ctx, obj: pd.DataFrame) -> None: ...; def load_input(self, ctx) -> pd.DataFrame: .... Key one bucket+prefix per resource instance; register under a stable name in Definitions.resources; set io_manager_key="..." on each @asset that should route through it. Snowflake and DuckDB have ready-made IO managers in dagster-snowflake-pandas and dagster-duckdb-pandas; parquet-on-S3 you often write yourself (~30 lines).
• AutomationCondition DSL. Atoms: on_cron(cron_expr), on_missing(), any_deps_updated(), all_deps_updated(), in_progress(), newly_updated(). Operators: & | ~. Modifiers: .since_last_handled(). Shortcuts: AutomationCondition.eager(), AutomationCondition.on_cron("0 * * * *"), AutomationCondition.on_missing(). Compose for expensive assets: (on_missing() | any_deps_updated().since_last_handled()) & ~in_progress(). Attach via automation_condition= on @asset; the runtime evaluator ticks every ~30 seconds and launches materializations for truthy (asset, partition) pairs.
• dagster-dbt deep integration. from dagster_dbt import DbtCliResource, dbt_assets; @dbt_assets(manifest="dbt_project/target/manifest.json", dagster_dbt_translator=MyTranslator()); def analytics_dbt_assets(context, dbt: DbtCliResource): yield from dbt.cli(["build"], context=context).stream(). Sub-class DagsterDbtTranslator to attach partitions (get_partition_mapping), groups (get_group_name), tags (get_tags), or asset keys (get_asset_key) per model. Regenerate manifest.json on every dbt change via CI; the Dagster code location picks up new/changed models on reload.
• Airflow migration checklist. (a) Deploy Dagster+ (or self-host with Postgres + S3), (b) install dagster-airflow for the bidirectional bridge, (c) migrate the dbt tier first via dagster-dbt (biggest observability win), (d) wire the Airflow → Dagster bridge for teams whose scheduling stays in Airflow, (e) migrate teams quarterly starting with analytics-adjacent, (f) leave operator-heavy long-tail on Airflow, bridged from Dagster, (g) decommission Airflow only when the last DAG is off or bridged. Every step has a rollback that is < 1 day per team.
• Definitions module structure. One Definitions(assets=[...], resources={...}, sensors=[...], schedules=[...], jobs=[...]) per code location. Multiple code locations for multi-team deployments (one per team); code locations can reference each other's assets via AssetKey. Resources are dependency-injected — SnowflakeResource(...), S3Resource(...), DbtCliResource(...) — and swappable in tests via Definitions.merge(...) or environment-based EnvVar(...).
• Asset check patterns. For every mart-tier asset, ship at least (a) a row-count-drift check comparing this partition to the previous one, and (b) a freshness check comparing max(updated_at) to now. Set blocking=True on both so a failure blocks downstream. Non-critical checks stay severity=WARN for observability without blocking. Every check emits metadata={...} so on-call can triage without opening logs.
• Run coordinator cost caps. In dagster.yaml under run_coordinator, use QueuedRunCoordinator with max_concurrent_runs for a global cap plus tag_concurrency_limits for per-tag caps (compute=snowflake at 5; tier=weekly at 1; tier=backfill at 5). Every expensive asset should carry a compute_kind= (surfaced in UI) and a tags={"compute_kind": "..."} (used by the coordinator). The two together bound both the visible cost and the enforced cost.
• Branch deployments contract. Use Dagster+ branch deployments as the default review artifact for any PR that adds or changes an asset. Provision on PR-open via GitHub webhook; scope storage to s3://bucket/branch-deployments/pr-<n>/; auto-teardown 24h after PR close. Reviewers click "materialise" in the PR-scoped UI to validate end-to-end before approving. For OSS Dagster teams, a shared staging deployment is the acceptable minimum but requires team discipline to prevent one team's test from breaking another's.
• Sensors, schedules, and Declarative Automation — one mechanism per firing reason. Sensors react to external events (S3 file, SQS message, out-of-Dagster completion); schedules run job-level cron; Declarative Automation drives asset-graph freshness. Never let two mechanisms target the same materialization or you'll get spurious double-launches. Tag every run with trigger=sensor|schedule|automation for observability.
• What senior interviewers score highest. Naming "software-defined asset" as the unit of work (not "task"); distinguishing asset-centric from task-centric in sentence one; naming Declarative Automation and AutomationCondition as the 1.6+ replacement for AutoMaterializePolicy; naming dagster-dbt deep integration (parses manifest.json, one asset per model); naming dagster-airflow as the migration bridge (not "big bang"); conceding Airflow's operator-library win as a legitimate reason to keep it for the long tail. These are the senior signals that separate architects who have designed a Dagster deployment from candidates who have only read the tutorial.

Frequently asked questions

What is Dagster in one sentence?

dagster is an asset-centric orchestrator where the unit of work is a software-defined asset — a Python-decorated function that declares a durable data object with a typed identity, upstream/downstream lineage, quality checks, an IO manager for storage, an optional partition scheme, and an optional automation condition — and the runtime computes which assets need to run to bring the asset graph into a desired state, launches those materializations, records the events in an append-only log, and surfaces lineage plus check status in a native asset-graph UI. Compared to Airflow's task-centric model (where you schedule computations), Dagster asks you to declare which data should exist and lets the runtime handle the scheduling. Every senior data-engineering interview in 2026 probes Dagster because the choice of orchestrator determines the mental model your entire platform hard-codes assumptions against for years — and because Dagster's dagster-dbt deep integration, per-partition backfills, and Declarative Automation are the three features task-centric orchestrators genuinely cannot replicate without a full rewrite.

Dagster vs Airflow — when do I pick each?

Default to Dagster when the platform is greenfield, the unit of work is naturally a dataset (dbt model, S3 partition, warehouse mart, feature table), and the team wants native data-lineage, per-partition backfills, and inline data-quality checks without bolting third-party tools onto a scheduler. Default to Airflow when the workload is operator-heavy (500+ community operators for SaaS APIs, cloud services, and legacy systems), the existing platform is 100+ DAGs deep (migration cost is real), or the team's mental model is genuinely task-first (batch jobs on cron, no dataset-shaped unit of work). The dagster vs airflow question is not "which one is better" — it is "which mental model does your platform want to describe itself in," and the honest answer is that a green-field analytics/ML team almost always picks Dagster while a hundred-DAG Airflow shop almost always keeps Airflow for the long tail even as they migrate hot paths to Dagster via dagster-airflow bridges. Concede the operator-library win in every interview answer — refusing to acknowledge Airflow's ecosystem strength is a junior signal.

What is a software-defined asset?

A dagster software defined assets (SDA) is a Python function decorated with @asset that declares (a) the asset's identity via an AssetKey — hierarchical, defaults to the function name; (b) its upstream dependencies via typed function parameters (def orders_clean(raw_orders: pd.DataFrame)) or an explicit deps=[...] list; (c) how to materialise the underlying data (the function body); and (d) optionally its partitions_def, automation_condition, check_specs, io_manager_key, group_name, compute_kind, tags, and code_version. The Dagster runtime uses this metadata to build the full asset graph, decide what to run when a materialization is requested (either manually, from a sensor, from a schedule, or from an automation condition), surface lineage in the UI, and route inputs/outputs through configured IO managers. Every senior Dagster interview starts with this construct because nothing else in Dagster makes sense without it — schedules, sensors, checks, backfills, automation, dagster-dbt, and IO managers all attach to or reference SDAs.

What is Declarative Automation vs Auto-Materialize?

dagster auto materialize was the older Dagster 0.x/1.0-1.5 policy engine — AutoMaterializePolicy.eager(), AutoMaterializePolicy.lazy(), FreshnessPolicy(...) — where the runtime auto-materialised assets based on a fixed set of policies. Declarative Automation is the Dagster 1.6+ replacement built around the composable AutomationCondition DSL: you attach an AutomationCondition to each asset (e.g. AutomationCondition.on_cron("0 * * * *") & AutomationCondition.any_deps_updated().since_last_handled()), the runtime evaluates the condition per asset partition every ~30 seconds, and launches materializations for any (asset, partition) pair whose condition is truthy. The DSL's atoms are on_cron, on_missing, any_deps_updated, all_deps_updated, in_progress, and newly_updated; operators are & | ~; modifiers are .since_last_handled() and friends. The old AutoMaterializePolicy shortcuts still exist as AutomationCondition.eager() and AutomationCondition.on_cron(...), so migration is largely non-breaking; new code should use the DSL directly because it composes cleanly (e.g. rate-limiting and back-pressure become one-line composites rather than custom sensors). Every senior interview in 2026 asks about this transition — knowing the new DSL by name is a senior signal.

How does Dagster integrate with dbt?

dagster-dbt provides deep integration — not "there's an operator." Wire it in with one decorator: @dbt_assets(manifest="dbt_project/target/manifest.json") parses the dbt manifest at Dagster code-load time and creates one Dagster asset per dbt model with correct lineage, tags, and per-model materialization events. When the underlying function runs dbt.cli(["build"], context=context).stream(), dbt executes and per-model events are translated into Dagster materialization events in real time — so the Dagster UI shows every model's start/success/failure as a distinct asset event, complete with row counts (from dbt's run_results.json), materialization type (view/table/incremental), and lineage into any downstream non-dbt Dagster asset that depends on it. Contrast with Airflow's DbtOperator which runs dbt build as one task (you never see individual models), or Astronomer Cosmos which improved this by creating one Airflow task per model but keeps a task-graph UI. Attach partitions via a DagsterDbtTranslator subclass that returns partitions_def=DailyPartitionsDefinition(...) for models tagged partitioned_daily; the runtime passes the partition key as a dbt variable at execution time. This is why the dbt tier is almost always the first thing migrated from Airflow to Dagster — the observability upgrade is dramatic and the migration is one decorator plus a translator subclass.

Is Dagster production-ready in 2026?

Yes for both OSS Dagster and Dagster+. OSS Dagster reached 1.0 in 2022 and has shipped stable releases on a rapid cadence since; the 1.6+ Declarative Automation release brought the last major abstraction into place. Production deployments in the thousands of assets are common; the largest known deployments are in the tens of thousands. Dagster+ (the managed offering that launched in 2024) provides branch deployments, Insights (query-level observability), managed compute, and single-tenant options; SLAs and support match what you'd expect from a serious managed data platform. The remaining rough edges in 2026 are largely around operations: (a) the self-hosted Postgres persistence layer needs the same care as any other stateful service (backups, replicas, tuning), (b) very-large-scale deployments (10k+ assets) benefit from careful code-location partitioning to keep code-load times bounded, (c) dagster-dbt requires the dbt manifest.json to be regenerated on every dbt change (a solved CI problem but one you must set up), and (d) the ecosystem of Dagster resources is thinner than Airflow's operator library — count on writing your own IO managers and resources for the long tail of SaaS integrations. None of these are blockers; they are the standard "run a stateful platform in production" concerns. The interview's question here is almost always "is it mature enough to bet on" — the answer in 2026 is unambiguously yes, with the caveat that you must run it operationally the way you would any other tier-1 orchestration platform.

Practice on PipeCode

• Drill the SQL practice library → for the mart-tier aggregate queries, MERGE-into-Snowflake incremental patterns, and window-function shapes senior interviewers use to test asset-driven analytics designs.
• Rehearse system design against the design practice library → for orchestrator-selection, Dagster-vs-Airflow migration, and multi-region partitioned-pipeline scenarios that mirror the senior Dagster design interview.
• Practise the aggregation practice library → for the fan-in / rollup / time-window aggregate shapes that dominate mart-tier and rollup assets in a production Dagster graph.
• Sharpen the streaming axis with the streaming practice library → for the sensor-driven, CDC-fed, per-file-triggered pipeline patterns that show up in senior orchestration interviews.
• Warm up on the ETL practice library → for the raw → clean → mart tiering that maps 1:1 onto the Dagster three-tier asset graph pattern used throughout this guide.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the software-defined-asset mental model, the four-axis partition matrix, and the AutomationCondition DSL against real graded inputs.

Lock in dagster muscle memory

Docs explain the API. PipeCode drills explain the decision — when to model a job as an asset graph instead of a task DAG, when a `TimeWindowPartitionMapping` beats a hand-rolled watermark, when Declarative Automation's `all_deps_updated()` gate is worth the extra composition, when the `dagster-dbt` deep integration is the wedge that justifies migrating a whole platform. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the asset-centric orchestration decisions senior data engineers actually face in 2026.

Practice SQL problems →
Practice design problems →