Construcción de activos de Dagster

¿Qué hace Construcción de activos de Dagster?

Cuándo usarlo

• Crear o refactorizar activos y trabajos definidos por software de Dagster.
• Modelar tablas/archivos/modelos ML como activos con linaje.
• Añadir particiones, rellenos, comprobaciones de activos, programaciones o sensores.
• NO usar para Airflow (usar las habilidades de Airflow).

Flujo de trabajo

- [ ] Modelar cada salida como un @asset; declarar dependencias mediante argumentos de función
- [ ] Añadir particiones para datos segmentados por tiempo o categoría
- [ ] Mover E/S (lecturas/escrituras) a gestores de E/S o recursos
- [ ] Añadir comprobaciones de activos para calidad de datos
- [ ] Programar/sensor para materializar

1. Piensa en activos, no en tareas. Un activo es un objeto persistente (una tabla, archivo, modelo). Declara dependencias referenciando activos upstream como parámetros de función: Dagster construye el grafo de linaje automáticamente.
2. Particiona activos que estén naturalmente segmentados (por día, región) para poder materializar/rellenar un segmento a la vez.
3. Recursos y gestores de E/S mantienen conexiones y lógica de lectura/escritura, manteniendo los cuerpos de los activos centrados en la transformación y haciéndolos comprobables.
4. Comprobaciones de activos adjuntan aserciones de calidad de datos a un activo.

Patrones

Activos particionados con dependencia:

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:  # depende 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})

Recurso / gestor de E/S — inyecta un cliente de almacén en lugar de construirlo dentro del activo:

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

Programa un trabajo particionado para que cada ejecución materialice la partición más reciente; usa un sensor para materializar cuando aparezca un archivo/activo upstream.

Errores comunes

• Pensar en tareas — usar @op/trabajos simples para todo pierde el linaje, la observabilidad y los rellenos conscientes de particiones que los activos proporcionan gratis.
• E/S dentro de los cuerpos de los activos — codificar conexiones dificulta la comprobación de los activos; usa recursos/gestores de E/S.
• Activos particionados no idempotentes — materializar una partición debe sobrescribir esa partición, no añadir (ver writing-idempotent-transformations).
• Omitir comprobaciones de activos — sin ellas, los datos incorrectos se materializan silenciosamente; Dagster muestra fallos de comprobación en la interfaz de usuario y puede bloquear downstream.
• Definiciones de partición no coincidentes entre activos dependientes — mantén partitions_def consistente para que los mapeos se resuelvan.