Construire des Assets Dagster

Que fait Construire des Assets Dagster ?

Quand l'utiliser

• Créer ou refactoriser des assets définis par logiciel et des jobs Dagster.
• Modéliser des tables/fichiers/modèles ML en tant qu'assets avec leur lignage.
• Ajouter des partitions, des backfills, des contrôles d'assets, des programmations ou des capteurs.
• Ne PAS utiliser pour Airflow (utilisez les compétences Airflow).

Workflow

- [ ] Modéliser chaque sortie comme un @asset ; déclarer les dépendances via les paramètres de fonction
- [ ] Ajouter des partitions pour les données découpées par temps ou catégorie
- [ ] Déplacer les E/S (lectures/écritures) dans des gestionnaires d'E/S ou des ressources
- [ ] Ajouter des contrôles d'assets pour la qualité des données
- [ ] Planifier ou ajouter un capteur pour matérialiser

1. Pensez en assets, pas en tâches. Un asset est un objet persistant (une table, un fichier, un modèle). Déclarez les dépendances en référençant les assets en amont comme paramètres de fonction — Dagster construit automatiquement le graphe de lignage.
2. Partitionnez les assets qui sont naturellement découpés (par jour, région) afin de pouvoir matérialiser ou faire un backfill d'une seule partition à la fois.
3. Les ressources et les gestionnaires d'E/S contiennent les connexions et la logique de lecture/écriture, ce qui permet de garder les corps d'assets concentrés sur la transformation et de les rendre testables.
4. Les contrôles d'assets attachent des assertions de qualité des données à un asset.

Patrons

Assets partitionnés avec une dépendance :

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:  # dépend de raw_orders
    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 / gestionnaire d'E/S — injectez un client d'entrepôt plutôt que de le construire à l'intérieur de l'asset :

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

Programmez un job partitionné pour que chaque exécution matérialise la dernière partition ; utilisez un capteur pour matérialiser quand un fichier/asset en amont apparaît.

Pièges courants

• Pensée en tâches — utiliser @op/jobs bruts pour tout perd le lignage, l'observabilité et les backfills tenant compte des partitions que les assets offrent gratuitement.
• E/S à l'intérieur des corps d'assets — coder en dur les connexions rend les assets non testables ; utilisez des ressources/gestionnaires d'E/S.
• Assets partitionnés non idempotents — matérialiser une partition doit écraser cette partition, pas ajouter (voir writing-idempotent-transformations).
• Ignorer les contrôles d'assets — sans eux, les mauvaises données se matérialisent silencieusement ; Dagster affiche les échecs de contrôle dans l'interface utilisateur et peut bloquer les étapes en aval.
• Définitions de partitions incompatibles entre assets dépendants — assurez-vous que le partitions_def est cohérent pour que les correspondances soient résolues.