Планування з урахуванням даних за допомогою Assets

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

Volker Janz

Senior Developer Advocate at Astronomer

Проблема з таймінгом

Конвеєр A завантажує сирі дані з API
Конвеєр B будує дашборд з цих даних
Якщо обидва працюють за cron, B може запуститись раніше, ніж A завершиться
Додавання часової «подушки» — ненадійно

Проблема з таймінгом

Що таке Asset?

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

Посилання на фрагмент даних з унікальною назвою
Коли це конкретна сутність даних, до активу можна додати URI
Це може бути файл, таблиця в БД або будь-які дані

Виробники й споживачі

Завдання можуть оновлювати активи, створюючи подію активу
Dags можна запускати при оновленні активів за допомогою asset schedule

Потік активів

Виробник: сигнал через 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