Construção de Assets Dagster

O que Construção de Assets Dagster faz?

Quando usar

• Criar ou refatorar assets e jobs definidos por software do Dagster.
• Modelar tabelas/arquivos/modelos de ML como assets com linhagem.
• Adicionar partições, backfills, verificações de assets, schedules ou sensors.
• Não usar para Airflow (use a skill de Airflow).

Fluxo de trabalho

- [ ] Modelar cada saída como um @asset; declarar deps via argumentos da função
- [ ] Adicionar partições para dados divididos por tempo/categoria
- [ ] Mover I/O (leituras/escritas) para IO managers ou recursos
- [ ] Adicionar verificações de assets para qualidade de dados
- [ ] Schedule/sensor para materializar

1. Pense em assets, não em tarefas. Um asset é um objeto persistente (uma tabela, arquivo, modelo). Declare dependências referenciando assets upstream como parâmetros da função — o Dagster constrói o grafo de linhagem automaticamente.
2. Particione assets que são naturalmente divididos (por dia, região) para que você possa materializar/backfill uma partição de cada vez.
3. Recursos e IO managers mantêm conexões e lógica de leitura/escrita, mantendo os corpos dos assets focados na transformação e tornando-os testáveis.
4. Verificações de assets anexam asserções de qualidade de dados a um asset.

Padrões

Assets particionados com dependência:

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 / IO manager — injete um cliente de warehouse em vez de construí-lo dentro do asset:

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

Schedule um job particionado para que cada execução materialize a partição mais recente; use um sensor para materializar quando um arquivo/asset upstream aparecer.

Armadilhas comuns

• Pensamento em tarefas — usar @op/jobs simples para tudo perde linhagem, observabilidade e backfills conscientes de partição que os assets oferecem gratuitamente.
• I/O dentro dos corpos dos assets — conectar-se diretamente torna os assets não testáveis; use recursos/IO managers.
• Assets particionados não idempotentes — materializar uma partição deve sobrescrever essa partição, não anexar (consulte writing-idempotent-transformations).
• Pular verificações de assets — sem elas, dados ruins são materializados silenciosamente; o Dagster exibe falhas de verificação na UI e pode bloquear downstream.
• Definições de partição incompatíveis entre assets dependentes — mantenha o partitions_def consistente para que os mapeamentos sejam resolvidos.