Dagster Assets erstellen

Was macht Dagster Assets erstellen?

Wann verwenden

• Erstellen oder Refactoren von Dagster-softwaredefinierten Assets und Jobs.
• Modellieren von Tabellen/Dateien/ML-Modellen als Assets mit Herkunftsnachweis (Lineage).
• Hinzufügen von Partitionen, Backfills, Asset-Checks, Zeitplänen oder Sensoren.
• NICHT für Airflow verwenden (nutzen Sie die Airflow-Skills).

Workflow

- [ ] Jeden Output als @asset modellieren; Abhängigkeiten über Funktionsparameter deklarieren
- [ ] Partitionen für zeit- oder kategoriebasierte Daten hinzufügen
- [ ] I/O (Lese-/Schreibzugriffe) in IO-Manager oder Ressourcen auslagern
- [ ] Asset-Checks für Datenqualität hinzufügen
- [ ] Zeitplan/Sensor zur Materialisierung

1. Denken Sie in Assets, nicht in Tasks. Ein Asset ist ein persistentes Objekt (Tabelle, Datei, Modell). Deklarieren Sie Abhängigkeiten, indem Sie Upstream-Assets als Funktionsparameter referenzieren – Dagster erstellt automatisch den Lineage-Graphen.
2. Partitionieren Sie Assets, die natürlich unterteilt sind (nach Tag, Region), damit Sie jeweils eine Partition materialisieren oder backfillen können.
3. Ressourcen und IO-Manager kapseln Verbindungen und Lese-/Schreiblogik, halten Asset-Bodies fokussiert auf die Transformation und machen sie testbar.
4. Asset-Checks fügen einem Asset Datenqualitätsprüfungen hinzu.

Patterns

Partitionierte Assets mit Abhängigkeit:

import dagster as dg

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

@dg.asset(partitions_def=daily)
def raw_orders(context: dg.AssetExecutionContext) -> None:
    day = context.partition_key
    write_parquet(f"raw/orders/{day}.parquet", fetch_orders(day))

@dg.asset(partitions_def=daily)
def orders_clean(context, raw_orders) -> None:  # hängt von raw_orders ab
    day = context.partition_key
    transform_and_load(day)

@dg.asset_check(asset=orders_clean)
def no_null_ids(context) -> dg.AssetCheckResult:
    n = count_null_order_ids(context.partition_key)
    return dg.AssetCheckResult(passed=n == 0, metadata={"null_ids": n})

Ressource / IO-Manager – injizieren Sie einen Warehouse-Client anstatt ihn im Asset zu konstruieren:

@dg.asset
def orders_summary(context, warehouse: WarehouseResource, orders_clean):
    warehouse.execute("insert into summary select ...")

Zeitplan für einen partitionierten Job – jeder Lauf materialisiert die neueste Partition; verwenden Sie einen Sensor, um zu materialisieren, sobald eine upstream-Datei/ein upstream-Asset erscheint.

Häufige Fehler

• Task-Denken – die Verwendung von bloßem @op/Jobs für alles verliert die Lineage, Beobachtbarkeit und die partition-bewussten Backfills, die Assets kostenlos bieten.
• I/O innerhalb von Asset-Bodies – hartkodierte Verbindungen machen Assets untestbar; nutzen Sie Ressourcen/IO-Manager.
• Nicht-idempotente partitionierte Assets – das Materialisieren einer Partition muss diese Partition überschreiben, nicht anhängen (siehe writing-idempotent-transformations).
• Asset-Checks auslassen – ohne sie werden schlechte Daten stillschweigend materialisiert; Dagster zeigt Check-Fehler in der UI an und kann Downstream blockieren.
• Nicht übereinstimmende Partitionsdefinitionen zwischen abhängigen Assets – halten Sie die partitions_def konsistent, damit die Zuordnungen aufgelöst werden.