Agendamento ciente de partições com Asset Partitions

Criando Pipelines de Dados com Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Quando ds não basta

{{ ds }} vincula cada execução a uma data, útil para Dags agendados por tempo
Mas em Dags acionados por asset, ds é None — não está disponível
O Dag downstream não sabe qual data acabou de ser atualizada
Asset Partitions propagam um partition_key do upstream para o downstream

Execução agendada upstream aciona execução downstream por asset, onde ds é None

CronPartitionTimetable

from airflow.sdk import dag, CronPartitionTimetable

@dag(schedule=CronPartitionTimetable("0 0 * * *", timezone="UTC"))
def sales_pipeline():
    ...

Toda execução agendada recebe um partition_key automaticamente
Execuções manuais não são particionadas sem fornecer a chave

Acessando chaves de partição

Em tarefas Python:

@task
def process_data(dag_run=None):
    partition_key = dag_run.partition_key
    print(f"Processing: {partition_key}")

Em templates SQL:

DELETE FROM daily_summary
WHERE order_date = '{{ dag_run.partition_key[:10] }}';

Disponível em todas as tarefas dentro de uma execução particionada

Eventos de asset particionados

from airflow.sdk import dag, task, Asset, CronPartitionTimetable

daily_sales = Asset("daily_sales")

@dag(schedule=CronPartitionTimetable("0 0 * * *", timezone="UTC"))
def sales_pipeline():
    @task(outlets=[daily_sales])
    def load_data(**context):
        ...

Outlet da tarefa + CronPartitionTimetable = evento de asset particionado
O evento carrega a mesma chave de partição da execução do Dag
Indica qual partição está pronta, não só que houve atualização

PartitionedAssetTimetable

from airflow.sdk import dag, task, Asset, PartitionedAssetTimetable

daily_sales = Asset("daily_sales")

@dag(schedule=PartitionedAssetTimetable(assets=daily_sales))
def sales_report():
    @task
    def generate_report(dag_run=None):
        partition_key = dag_run.partition_key
        print(f"Report for: {partition_key}")

Dispara apenas em eventos de asset particionados
A execução downstream herda a chave de partição

StartOfDayMapper

from airflow.sdk import (
    dag, PartitionedAssetTimetable,
    StartOfDayMapper,
)

@dag(schedule=PartitionedAssetTimetable(
    assets=daily_sales,
    partition_mapper_config={
        daily_sales: StartOfDayMapper()
    },
))
def sales_report():
    ...

Antes do mapeamento: 2026-04-23T00:00:00

Após StartOfDayMapper: 2026-04-23

Chaves temporais: StartOfHourMapper, StartOfWeekMapper, StartOfMonthMapper
Chaves não temporais: AllowedKeyMapper para regiões, departamentos

O fluxo completo

Fluxo completo de partição

Upstream: CronPartitionTimetable anexa a chave de partição a cada execução
A tarefa outlet emite um evento de asset particionado com essa chave
StartOfDayMapper normaliza o timestamp para uma data
Downstream: PartitionedAssetTimetable dispara e herda a chave mapeada

Vamos praticar!

Criando Pipelines de Dados com Airflow