Планування з урахуванням розділів даних із Asset Partitions

Створення конвеєрів даних з Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Коли ds недостатньо

{{ ds }} прив'язує запуск до дати — підходить для за розкладом Dags
Але в актив-тригерних Dags ds має значення None, його просто немає
Нижній Dag не знає, яка дата щойно оновилась
Asset Partitions передають partition_key з апстріму в даунстрім

Запуск за розкладом апстріму тригерить даунстрім за активом, де ds = None

CronPartitionTimetable

from airflow.sdk import dag, CronPartitionTimetable

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

Кожний запуск за розкладом автоматично отримує partition_key
Ручні запуски не розділяються, якщо ключ не задано

Доступ до ключів розділів

У Python-завданнях:

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

У SQL-шаблонах:

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

Доступно в кожному завданні в межах розділеного запуску

Події активів із розділами

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 завдання + CronPartitionTimetable = подія активу з розділом
Подія несе той самий ключ розділу, що й запуск Dag
Сигналізує, який розділ готовий, а не лише що дані оновлено

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}")

Тригериться лише на подіях активів із розділами
Даунстрімний запуск успадковує ключ розділу

StartOfDayMapper

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

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

До мапінгу: 2026-04-23T00:00:00

Після StartOfDayMapper: 2026-04-23

Часові ключі: StartOfHourMapper, StartOfWeekMapper, StartOfMonthMapper
Нечасові ключі: AllowedKeyMapper для регіонів, відділів

Повний потік

Повний потік розділів

Апстрім: CronPartitionTimetable додає ключ розділу до кожного запуску
Outlet-завдання генерує подію активу з розділом із цим ключем
StartOfDayMapper нормалізує позначку часу до дати
Даунстрім: PartitionedAssetTimetable тригериться та успадковує мапований ключ

Давайте потренуємось!

Створення конвеєрів даних з Airflow