Agendamento com reconhecimento de dados usando Assets

Criando Pipelines de Dados com Airflow

Volker Janz

Senior Developer Advocate at Astronomer

O problema de timing

Pipeline A carrega dados brutos de uma API
Pipeline B cria um dashboard a partir desses dados
Se ambos rodam em cron, B pode rodar antes de A terminar
Adicionar uma folga de tempo é frágil

O problema de timing

O que é um Asset?

from airflow.sdk import Asset
sales_data = Asset("s3://bucket/sales/daily.parquet")

Uma referência a um dado, identificada por um nome único
Uma URI pode ser associada quando o asset representa um dado concreto
Pode ser um arquivo, uma tabela de banco, ou qualquer dado

Produtores e consumidores

Tarefas podem atualizar assets, gerando um evento de asset
DAGs podem ser acionados em atualizações via um asset schedule

Fluxo de assets

Produtor: sinalizando com outlets

sales_data = Asset("s3://bucket/sales/daily.parquet")


@task(outlets=[sales_data])
def write_sales():
    # Write data to S3
    ...

Com sucesso, o Airflow registra que o asset foi atualizado
O produtor não precisa saber sobre nenhum consumidor

Consumidor: agendando por um asset

sales_data = Asset("s3://bucket/sales/daily.parquet")


@dag(schedule=[sales_data])
def build_dashboard():
    ...

Sem expressão cron, só a referência ao asset
O Airflow aciona o consumidor automaticamente após a atualização

Agendamento condicional

sales = Asset("s3://bucket/sales.parquet")
inventory = Asset("s3://bucket/inventory.parquet")


# Esperar AMBOS os assets atualizarem
@dag(schedule=(sales & inventory))
def full_report(): ...


# Acionar quando QUALQUER asset atualizar
@dag(schedule=(sales | inventory))
def quick_refresh(): ...

Verificando assets com a CLI

$ airflow assets list

name                                  | group    | uri
s3://data-lake/sales/daily.csv        | asset    | s3://data-lake/sales/daily.csv

$ airflow assets details --name "s3://data-lake/sales/daily.csv"

property_name    | property_value
=================+=====================                                         
name             | s3://data-lake/sales/daily.csv
                ...
updated_at       | 2026-04-14T10:00:50.820041Z

airflow assets list mostra todos os assets registrados
airflow assets details mostra metadados, incluindo updated_at

Vamos praticar!

Criando Pipelines de Dados com Airflow