Планирование на основе ресурсов данных

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

Volker Janz

Senior Developer Advocate at Astronomer

Проблема синхронизации

Pipeline A загружает сырые данные из API
Pipeline B строит дашборд на основе этих данных
Если оба работают по cron, B может запуститься до завершения A
Добавление временного буфера — ненадёжное решение

Проблема синхронизации

Что такое ресурс данных (Asset)?

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

Ссылка на единицу данных, идентифицируемая уникальным именем
URI прикрепляется к ресурсу, если он представляет конкретный объект данных
Это может быть файл, таблица базы данных или любые другие данные

Производители и потребители

Задачи могут обновлять ресурсы, создавая событие ресурса
DAG-и можно запускать по обновлению ресурса с помощью расписания на основе ресурсов

Поток ресурсов данных

Производитель: сигнализация через outlets

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


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

При успешном завершении Airflow фиксирует обновление ресурса
Производителю не нужно знать ни о каком потребителе

Потребитель: планирование на основе ресурса

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


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

Никакого cron-выражения — только ссылка на ресурс
Airflow запускает потребителя автоматически после обновления

Условное планирование

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


# Wait for BOTH assets to update
@dag(schedule=(sales & inventory))
def full_report(): ...


# Trigger when ANY asset updates
@dag(schedule=(sales | inventory))
def quick_refresh(): ...

Проверка ресурсов через 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 выводит все зарегистрированные ресурсы
airflow assets details показывает метаданные, включая updated_at

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

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