Организация сложных DAG с помощью групп задач

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

Volker Janz

Senior Developer Advocate at Astronomer

Проблема сложности

Сложный DAG

Большие DAG сложно читать в представлении графа
Названия задач сливаются
Новым участникам команды трудно найти свои задачи

@task_group

from airflow.sdk import dag, task, task_group

@task_group(
    group_id="ingest_orders",
    default_args={"retries": 3},
)

def process_orders():
    @task
    def extract_orders():
        return [{"id": 1, "amount": 99.99}]

    @task
    def transform_orders(orders):
        return [{"id": o["id"], "total": o["amount"] * 1.08}
                for o in orders]

    return transform_orders(extract_orders())

group_id задаёт пользовательский идентификатор, по умолчанию равный имени функции
default_args применяется ко всем задачам группы и исключает дублирование настроек
Группы задач отображаются в интерфейсе как раскрываемые блоки

Как это выглядит в интерфейсе Airflow

Без групп задач

Простой граф с тремя задачами в ряд: extract_orders, transform_orders, load — все на одном уровне

Все задачи на одном уровне

С группами задач

Граф со свёрнутым блоком process_orders, содержащим extract и transform, за которым следует задача load вне группы

Сворачиваемые блоки в интерфейсе

Вложенность и пользовательские отображаемые имена

@task_group(group_display_name="Process All")
def process_all():

    @task_group(group_display_name="Ingest Orders")
    def orders():
        return transform(extract())

    @task_group(group_display_name="Process Returns")
    def returns():
        return transform(extract())

    return {
        "orders": orders(),
        "returns": returns(),
    }

Группы задач можно вкладывать для сложных пайплайнов
group_display_name задаёт читаемую метку в интерфейсе (поддерживаются даже эмодзи)

Фабричный паттерн

@task_group
def process_source(source_name, source_path):

    @task
    def extract():
        return read_data(source_path)

    @task
    def transform(data):
        return clean_data(data)

    return transform(extract())


# Reuse the same pattern for different sources
orders = process_source("orders", "/data/orders.csv")
returns = process_source("returns", "/data/returns.csv")
events = process_source("events", "/data/events.csv")

Группы задач — это декорированные функции Python
Их можно вызывать несколько раз с разными параметрами
Фабричный паттерн позволяет повторно использовать логику пайплайна

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