Планирование с учётом разделов Asset Partitions

Построение пайплайнов данных с Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Когда ds недостаточно

{{ ds }} привязывает каждый запуск к дате — это работает для запланированных DAG
В asset-triggered DAG значение ds равно None и недоступно
Дочерний DAG не знает, за какую дату были обновлены данные
Asset Partitions передают partition_key от родительского DAG к дочернему

Запланированный запуск upstream запускает downstream DAG, где 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] }}';

Доступно в каждой задаче разбитого на разделы запуска

События asset с разделами

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 = событие asset с разделом
Событие несёт тот же ключ раздела, что и запуск 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}")

Запускается только по событиям asset с разделом
Дочерний запуск наследует ключ раздела

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 для регионов, подразделений

Полный поток

Полный поток обработки разделов

Upstream: CronPartitionTimetable прикрепляет ключ раздела к каждому запуску
Задача outlet генерирует событие asset с разделом и этим ключом
StartOfDayMapper нормализует метку времени до даты
Downstream: PartitionedAssetTimetable запускается и наследует преобразованный ключ

Давайте потренируемся!

Построение пайплайнов данных с Airflow