TaskFlow API

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

Volker Janz

Senior Developer Advocate at Astronomer

Класичний підхід

def extract_data():
    return {"users": 150, "events": 4200}


with DAG("etl_pipeline") as dag:
    t1 = PythonOperator(task_id="extract",
                        python_callable=extract_data)
    t2 = PythonOperator(task_id="summary",
                        python_callable=print_summary)

    t1 >> t2

Підхід TaskFlow

from airflow.sdk import dag, task


@dag
def etl_pipeline():
    @task
    def extract_data():
        return {"users": 150}


    data = extract_data()
    print_summary(data)

Чому TaskFlow?

Менше шаблонного коду: декоратори замінюють екземпляри операторів
Неявні залежності: повернені значення автоматично з'єднують задачі
Читабельність: Dag читається як звичайний Python-скрипт
Класичні оператори лишаються для інтеграцій провайдерів

TaskFlow API - visual

Airflow UI: Grid view

Airflow Grid view

Кожна колонка — це Dag run
Кожен рядок — це задача
Кольори показують статус: green = успіх, red = помилка
Натисніть на квадрат, щоб переглянути логи та деталі

Airflow UI: Graph view

Airflow Grid view

Фокусується на одному Dag run
Показує залежності та паралельне виконання
Налаштуйте деталі в параметрах

Airflow UI: версіонування Dag

Dag version indicator in Airflow UI

Автоматично відстежує структурні зміни
Кожен запуск прив'язаний до активної на той час версії
Версії дивіться в панелі Dag details

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

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