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 — визуализация

Интерфейс Airflow: представление Grid

Представление Grid в Airflow

Каждый столбец — это запуск DAG
Каждая строка — это задача
Цвета обозначают статус: зелёный = успех, красный = ошибка
Нажмите на ячейку, чтобы просмотреть журналы и детали

Интерфейс Airflow: представление Graph

Представление Grid в Airflow

Фокус на одном запуске DAG
Показывает зависимости и параллельное выполнение
Дополнительные параметры — в настройках

Интерфейс Airflow: версионирование DAG

Индикатор версий DAG в интерфейсе Airflow

Автоматически отслеживает структурные изменения
Каждый запуск привязан к активной на тот момент версии
Версии можно найти в панели сведений о DAG

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

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