DAG, задачи и операторы

В прошлом уроке мы увидели файл пайплайна «Гонца» целиком, но без объяснений. Там были DAG, @task и стрелки >>. Сегодня разберёмся, из чего Airflow собирает пайплайн и как шаги передают друг другу данные. Заодно пайплайн подрастёт: Отто попросил добавить в отчёт курьеров, а их удобно выгружать параллельно с заказами.

План ремонта

Ремонт квартиры состоит из работ. Нельзя красить стены, пока их не оштукатурили. Нельзя класть плитку в ванной, пока не провели трубы. А электрика и сантехника друг от друга не зависят, их можно делать одновременно двумя бригадами.

Если нарисовать работы кружками, а «после чего» стрелками, получится схема, по которой прораб каждое утро решает, кому что делать. В ней есть одно правило: стрелки не могут образовать круг. Если покраска после штукатурки, а штукатурка после покраски, ремонт никогда не начнётся.

Такая схема и есть DAG.

DAG

DAG расшифровывается как directed acyclic graph, направленный ациклический граф. Разберём по словам:

• граф: задачи и связи между ними;
• направленный: у каждой связи есть «до» и «после», чистка идёт после выгрузки, а не наоборот;
• ациклический: нельзя вернуться в задачу, с которой начал. Иначе пайплайн не знал бы, с чего начать.

В Airflow DAG это весь пайплайн: какие в нём шаги, в каком порядке и по какому расписанию. Один файл на Python может описать один или несколько DAG.

Задачи и операторы

Задача это один шаг DAG с именем task_id. Выгрузка заказов, чистка, загрузка, проверка: всё это задачи.

Оператор это шаблон: какую работу задача умеет делать. PythonOperator вызывает функцию на Python, BashOperator выполняет команду в терминале, EmptyOperator ничего не делает и служит точкой сбора. Задача это оператор с конкретными настройками и своим task_id.

Вот одна и та же задача, написанная двумя способами:

from airflow.providers.standard.operators.python import PythonOperator

def extract_orders():
    ...

PythonOperator(task_id="extract_orders", python_callable=extract_orders)

from airflow.sdk import task

@task
def extract_orders():
    ...

extract_orders()

Второй способ называется TaskFlow. Декоратор @task превращает функцию в задачу, а вызов функции внутри DAG добавляет задачу в граф. Имя задачи берётся из имени функции. Для задач на Python TaskFlow короче, им и будем пользоваться. Операторы понадобятся, когда работа не на Python: SQL в хранилище, команда в терминале, файл в S3.