airflow-manual/educational-tasks.md

Практика по Apache Airflow

Здесь мы знакомимся с Airflow на готовых примерах: запускаем DAG, меняем задачи и смотрим, что произошло. Свой цельный процесс обработки данных вы соберете в курсовых работах дальше по программе.

Как проходить практику

В каждом блоке сначала выполните основную практику. Задания "По желанию" находятся в конце блока: их можно пропустить и перейти дальше. Сложность указана отдельно. Если дополнительному заданию нужны результаты другого упражнения, это написано в его начале.

Основные задания одного блока выполняйте последовательно, сохраняя изменения. Перед новым блоком прочитайте условия начала. Загрузка CSV и проверки DQ используют одну таблицу, поэтому идут вместе.

Блок	Примеры	Что наблюдаем	Перед практикой
1. Первый DAG	hello_world_dag.py	Задачи, зависимости, операторы, расписание	Главы 4 и 5; для 1.3 - расписания
2. SQL и подключения	sql_basic_dag.py	Connection, SQL, ошибка подключения	Подключения
3. Работа с файлами	file_operations_dag.py	Результаты отдельных задач	Главы 5 и 6
4. Загрузка и DQ	csv_to_postgres.py, csv_to_postgres_dq.py	XCom, данные в БД, ошибка проверки	Глава 6, шаблоны и XCom
5. Разбор ETL	data_processing_dag.py	Параллельные шаги и путь данных	Глава 5
6. Ветвление	branching_dag.py	Variables, выбранная ветка, skipped	Глава 6 и Variables
7. Ошибки и повторы	error_handling_dag.py	Попытки, обработчики, итог запуска	Глава 6; для 7.B - callback
8. Пулы и группировка	resource_management_dag.py	Слоты, XCom, TaskGroup	Пулы, XCom и TaskGroup

Главы доступны в оглавлении учебника. Инструкция запуска и параметры учебной БД находятся в README стенда, файлы примеров - в airflow-docker/dags/.

Откройте http://localhost:8080, войдите как admin / admin. Для выбранного DAG включите переключатель и нажмите Trigger DAG. Дождитесь завершения запуска, затем откройте его граф и логи задач. После правки Python-файла дождитесь обновления вкладки Code; новый запуск должен использовать новый код. Если DAG исчез или перестал обновляться, проверьте Import Errors и логи scheduler.

Каталог /opt/airflow/data/ в контейнере соответствует airflow-docker/data/ на компьютере. Команды ниже выполняются из airflow-docker. Подойдет терминал Linux или PowerShell в Windows.

Для SQL можно открыть консоль учебной БД:

docker compose exec postgres-training psql -U student -d training

В ней выполняются SQL-запросы из заданий; выход - \q. Сохраняйте исходные DAG перед опытами вне каталога dags: две Python-копии с одинаковым dag_id мешают Airflow выбрать нужную.

1. Первый DAG

Начните с исходного hello_world_dag.py. В нем три задачи: start_task, date_task, end_task. Включение DAG с расписанием может создать автоматический запуск; для опытов выбирайте свой ручной запуск.

1.1. Добавить задачу

Основная практика. Сложность: начальная.

Запустите исходный DAG и найдите сообщения всех трех задач. Переведите приветствие на русский, затем добавьте функцию с фиктивной погодой и PythonOperator с task_id="weather_task". Включите ее между start_task и date_task, изменив существующую цепочку зависимостей.

Что должно получиться: в новом запуске четыре задачи со статусом success, а в логе weather_task есть выбранный вами текст. В графе видно, какая задача должна закончиться перед ней.

1.2. Создать файл через BashOperator

Основная практика. Сложность: начальная.

Добавьте между date_task и end_task задачу write_note. Импорт BashOperator уже есть. В bash_command запишите команду echo 'Airflow works' > /opt/airflow/data/output/note.txt.

После запуска проверьте файл:

docker compose exec airflow-webserver cat /opt/airflow/data/output/note.txt

Что должно получиться: пять успешных задач и файл с текстом Airflow works. При повторном запуске этот файл перезаписывается.

1.3. Посмотреть расписание

Основная практика. Сложность: начальная.

Замените schedule_interval на '0 0 * * 1'. Найдите новое расписание и следующий автоматический запуск в UI. Сопоставьте дату запуска с часовым поясом DAG. catchup=False оставьте как есть.

Что должно получиться: UI показывает запуск по понедельникам в 00:00 в часовом поясе DAG. Ручной запуск по кнопке по-прежнему выполняется сразу, когда доступны ресурсы. Для проверки расписания ждать понедельника не нужно. После опыта выключите переключатель DAG.

1.A. Ограничить одновременные запуски

По желанию. Сложность: средняя.

В DAG(...) задайте max_active_runs=1. В одну Python-функцию временно добавьте time.sleep(30) с импортом time. Включите DAG и запустите его вручную дважды с разными Run Id.

Что должно получиться: пока первый запуск выполняет задержку, второй не начинает свои задачи. После первого он продолжает работу. Удалите задержку и выключите DAG; ограничение можно оставить.

2. SQL и подключения

Используйте исходный sql_basic_dag.py. Он создает public.students_sample, очищает ее перед вставкой и добавляет трех студентов. После выполнения данные остаются для проверки. Повторный полный запуск снова оставит три строки. Остальные таблицы учебной БД этот DAG не меняет.

2.1. Подключиться к учебной БД

Основная практика. Сложность: начальная.

Запустите DAG с готовым соединением postgres_training. В Admin > Connections создайте student_postgres: Connection Type = Postgres, Host = postgres-training, Database/Schema = training, Login = student, Password = student, Port = 5432. Замените postgres_conn_id во всех SQL-задачах на свой Conn Id.

Проверьте таблицу:

SELECT name, age FROM public.students_sample ORDER BY name;

Что должно получиться: Alice - 22, Bob - 24, Charlie - 21. Новый запуск с вашим Connection также успешен.

Чтобы увидеть ошибку подключения, временно укажите несуществующий Conn Id и поставьте retries=0 в default_args. Запустите DAG и найдите причину в логе первой задачи. Исправьте Conn Id, верните исходное число повторов и создайте новый запуск. Ручная отметка задачи как успешной не проверяет подключение.

2.2. Добавить SQL-задачу

Основная практика. Сложность: средняя.

Добавьте PostgresOperator с task_id="update_student" между insert_data и query_data. Его SQL должен установить Alice возраст 25. Используйте то же соединение, что и остальные задачи.

Что должно получиться: после запуска у Alice возраст 25, остальные строки не изменились. Повторный полный запуск дает те же три строки: сначала DAG очищает таблицу, потом вставляет данные и выполняет UPDATE.

2.A. Добавить колонку

По желанию. Сложность: средняя. Начните после 2.2.

Добавьте email VARCHAR(100) в create_table_sql и заполните его в insert_data_sql. Для уже созданной таблицы сначала выполните:

ALTER TABLE public.students_sample ADD COLUMN email VARCHAR(100);

Команда нужна один раз. CREATE TABLE IF NOT EXISTS не меняет существующую схему, а TRUNCATE очищает только строки.

Что должно получиться: после запуска в таблице три строки с непустым email. Задание 2.B можно выполнять и без этой колонки. Удаление таблицы или всей базы для опыта не требуется.

2.B. Сохранить отчет

По желанию. Сложность: средняя. Требуется 2.2; задание 2.A можно пропустить.

После query_data добавьте Python-задачу. Через PostgresHook получите количество студентов, средний возраст и количество студентов каждого возраста. Запишите результат в /opt/airflow/data/output/students_report.txt. У хука есть методы get_first() и get_records(); они позволяют обойтись без ручной работы с курсором.

Что должно получиться: в файле 3 студента, средний возраст около 23.33 и по одному студенту возрастов 21, 24 и 25. Эти числа должны совпасть с SQL по students_sample.

3. Работа с файлами

Начните с исходного file_operations_dag.py. Он сам создает 100 строк в input/sample_data.csv; данные каждого полного запуска могут отличаться. Чужие входные файлы для этого блока не нужны.

3.1. Изменить преобразование

Основная практика. Сложность: начальная.

Запустите DAG, откройте data/output/processed_data.csv и summary.txt. Найдите в transform_data() правило вычисления salary_category. Измените порог категории High с 70000 на 80000, оставив границу Medium равной 50000. Запустите DAG еще раз.

Что должно получиться: в обработанном CSV 100 строк, у каждой строки с зарплатой от 80000 категория High. Сумма количеств High, Medium и Low в отчете равна 100. Сверяйте отчет с CSV того же запуска: случайные данные не позволяют сравнивать только числа из двух отчетов.

3.A. Добавить отдел и опыт

По желанию. Сложность: средняя.

В generate_sample_data() добавьте department, experience_years, education_level. Отдел выбирайте из sales, support, engineering; опыт должен быть от 0 до age - 18. Дополните read_and_validate_data() проверками новых полей. В transform_data() сохраните сотрудников с опытом больше трех лет в отдельный output/experienced.csv, сохранив полный processed_data.csv.

Что должно получиться: основной файл по-прежнему содержит 100 строк, опыт у всех допустимый, а в experienced.csv нет опыта 3 года и меньше.

3.B. Отчет по отделам

По желанию. Сложность: средняя. Требуется 3.A.

Добавьте задачу после write_summary. Она читает полный processed_data.csv и пишет output/departments.json: для каждого отдела количество сотрудников и среднюю зарплату. Для расчета подойдет pandas groupby, для записи - to_json.

Что должно получиться: сумма численности отделов равна 100, средние зарплаты совпадают с группировкой исходного CSV. График для этого задания не требуется.

3.C. Разделить обработку по отделам

По желанию. Сложность: высокая. Требуется 3.A, отчет 3.B необязателен.

После transform_data поставьте три параллельные задачи. Каждая читает полный processed_data.csv, отбирает свой отдел, проверяет уникальность id и сохраняет отдельный файл. После них добавьте сборку этих трех файлов и проверьте количество строк и уникальность id уже в общем результате. Не перезаписывайте исходный файл из параллельных задач.

Для опыта с повтором одна из задач должна намеренно падать только на первой попытке; используйте ti.try_number, retries=1 и retry_delay=timedelta(seconds=2). Этот прием разобран в блоке 7.

Что должно получиться: объединенный результат содержит те же 100 уникальных id. В графе три задачи имеют общего предшественника, но не зависят друг от друга. Повтор не добавляет дубликаты: файл отдела перезаписывается.

4. Загрузка CSV и проверка качества

Здесь работают два DAG и одна таблица public.orders. Начните с исходных csv_to_postgres.py и csv_to_postgres_dq.py. SQL-блок использовал другую таблицу. Сначала выполняйте основные задания, затем расширяйте схему по желанию.

4.1. Проследить загрузку и XCom

Основная практика. Сложность: начальная.

Запустите csv_to_postgres. В задаче generate_csv откройте вкладку XCom с ключом return_value: там путь к CSV. Найдите этот файл в data/output. Затем выберите load_csv_to_postgres, откройте Details и прокрутите до Rendered Templates. В opKwargs вместо шаблона параметра csv_path появился путь к файлу. Часть пути в UI может выглядеть как ***; сравните имя CSV с XCom.

Проверьте результат в БД:

SELECT COUNT(*) FROM public.orders;
SELECT order_id, amount FROM public.orders ORDER BY order_id LIMIT 5;

Что должно получиться: первый запуск на пустой таблице загружает 1000 строк. В XCom передается путь к файлу, сами строки хранятся в CSV и PostgreSQL. Каждый новый полный запуск генерирует новые заказы; после двух последовательных запусков на изначально пустой таблице будет 2000 строк. Это отличается от повторного выполнения только задачи загрузки с тем же CSV.

4.2. Добавить проверку и увидеть ошибку

Основная практика. Сложность: средняя.

После успешной загрузки отдельно запустите csv_to_postgres_dq. Автоматической связи между этими двумя DAG нет.

По образцу _check_has_rows() добавьте функцию и Python-задачу check_positive_amount перед data_quality_summary. Проверьте число строк, где amount <= 0, и бросьте ValueError, если такие строки есть. Новую задачу включите в существующую цепочку.

Что должно получиться на корректных данных: все задачи и DAG Run успешны, сводка перечисляет результаты проверок.

Для отрицательного сценария поменяйте знак суммы одного заказа:

UPDATE public.orders SET amount = -amount
WHERE order_id = (SELECT MIN(order_id) FROM public.orders);

Запустите только DQ-DAG. Новая проверка должна получить failed, сводка - upstream_failed, весь DAG Run - failed. В ее логе должно быть понятно, какое условие нарушено. У DQ-задач нет повторов: повторный запуск с теми же неверными данными не исправляет ошибку.

Восстановите сумму тем же UPDATE еще один раз и повторите DQ. Дождитесь успешного результата перед продолжением. Во время опыта не запускайте загрузчик и не удаляйте строки: восстановление относится к тому же минимальному order_id.

4.A. Провести новое поле через загрузку

По желанию. Сложность: высокая. Требуется завершенный основной блок 4.

Добавьте status со значениями NEW, PROCESSING, COMPLETED. Поле нужно провести через четыре участка: генератор CSV, DDL целевой таблицы, список колонок COPY и обе части итогового INSERT ... SELECT в _load_csv(). Во всех списках добавляйте его в конец. Порядок колонок CSV должен совпадать с COPY: CSV HEADER пропускает первую строку, но не сопоставляет поля по ее именам.

В DDL используйте status VARCHAR(20) NOT NULL DEFAULT 'NEW'. Существующую таблицу подготовьте один раз:

ALTER TABLE public.orders
ADD COLUMN status VARCHAR(20) NOT NULL DEFAULT 'NEW';

В EXPECTED_ORDERS_SCHEMA DQ-DAG добавьте в конец ("status", "character varying"). После загрузки проверьте значения status в БД и выполните DQ.

Что должно получиться: новые заказы содержат статусы из CSV, старые получили NEW, проверка схемы проходит. Проверяйте значения, а не только успешность задачи: забытый status в INSERT может оставить всем строкам значение по умолчанию.

Завершив опыт, удалите status из генератора, DDL, COPY и обеих частей INSERT ... SELECT. Уберите его и из EXPECTED_ORDERS_SCHEMA в DQ. Затем выполните ALTER TABLE public.orders DROP COLUMN status; и запустите DQ. Проверку положительной суммы из 4.2 сохраните.

4.B. Заменить PythonOperator на PostgresOperator

По желанию. Сложность: средняя. Используйте базовую схему после 4.2 или после восстановления в 4.A.

Замените задачу create_orders_table на PostgresOperator с тем же task_id, SQL создания таблицы и соединением postgres_training. Образец оператора есть в SQL-блоке. _get_conn() оставьте: загрузчик продолжает использовать эту функцию.

Что должно получиться: после замены DAG загружает новую порцию, затем DQ проходит. Имя задачи и цепочка зависимостей сохраняются.

4.C. Выполнить все проверки после ошибки

По желанию. Сложность: высокая. Требуется 4.2.

Временно замените ожидаемый тип amount в EXPECTED_ORDERS_SCHEMA на integer. Сначала запустите исходную цепочку: check_orders_schema упадет, последующие задачи получат upstream_failed.

У задач после проверки схемы, включая сводку, задайте trigger_rule="all_done". Повторите запуск. Прочитайте _log_dq_summary(): она собирает состояния всех проверок и сама падает, если хотя бы одна неуспешна.

Что должно получиться: проверки строк, дубликатов и положительной суммы выполняются даже после ошибки схемы. Сводка и DAG Run остаются failed. Успешная завершающая задача с all_done могла бы скрыть отказ внутри DAG; именно поэтому сводка проверяет результат.

Верните тип numeric и исходные trigger_rule, затем выполните успешный DQ-запуск.

5. Разбор ETL-примера

Начните с исходного data_processing_dag.py. Он сам создает свои customers.csv и orders.csv; таблица public.orders из блока 4 ему не нужна. Шаг simulate_load показывает подготовку к загрузке и печатает пример DDL. Записи в БД в этом DAG нет.

5.1. Проследить путь данных

Основная практика. Сложность: начальная.

Запустите DAG. Найдите две задачи извлечения и их файлы output/extracted_customers.csv, output/extracted_orders.csv. Проверьте, что transform_data() читает именно их. В графе обе задачи извлечения могут выполняться параллельно, а преобразование ждет обе.

В extract_orders() временно оставьте только заказы с amount >= 100 перед сохранением файла. Запустите DAG снова и сравните отчет с результатом до изменения.

Что должно получиться: исходный отчет содержит 6 заказов с выручкой 2105; после фильтра - 4 заказа с выручкой 2000. Лог simulate_load явно сообщает об имитации. Верните фильтр к исходному состоянию и запустите DAG перед дополнительными заданиями.

5.A. Добавить справочник продуктов

По желанию. Сложность: средняя.

В create_sample_data() добавьте input/products.csv с полями product, category, weight_kg. Для каждого из шести продуктов в заказах должна быть одна строка справочника. Добавьте параллельную задачу extract_products, сохраняющую output/extracted_products.csv, и подключите ее к transform_data. Объедините извлеченные заказы со справочником по product.

Что должно получиться: в результате 6 заказов, у всех заполнена категория, общая выручка осталась 2105. Если JOIN размножил строки или потерял заказы, проверьте уникальность и полноту справочника.

5.B. Расширить отчет

По желанию. Сложность: средняя. Справочник из 5.A не нужен.

В generate_report() добавьте топ-3 заказа по сумме, выручку по месяцам и среднюю сумму заказа каждого клиента. Сохраните output/detailed_report.txt.

Что должно получиться: топ-3 суммы - 1200, 400 и 300; все заказы относятся к октябрю 2023 года; средняя сумма заказа Alice Johnson равна 640. У исходного набора всего шесть строк, поэтому результат можно сверить вручную.

5.C. Валидация и архив

По желанию. Сложность: высокая.

Между преобразованием и simulate_load добавьте проверку: результат непустой, customer_id и name заполнены, суммы положительны. Для этой задачи задайте retry_delay=timedelta(seconds=2), сохранив один повтор. В create_sample_data() временно укажите для одного заказа несуществующий customer_id. Настройте on_failure_callback, который пишет в лог идентификатор задачи и причину ошибки; образец есть в задании 7.B.

После generate_report добавьте копирование transformed_data.csv в архивный файл с датой и временем в имени. Для этого подойдет shutil.copy.

Что должно получиться: после исчерпания повторов проверка падает, последующие шаги получают upstream_failed, новый архив не появляется. После возврата клиента к исходному значению весь DAG успешен, а содержимое нового архива совпадает с результатом запуска. Сообщение callback этой проверки находится в логе ее последней попытки.

6. Ветвление

Используйте исходный branching_dag.py. Ветки пока имитируют обработку, входные файлы не нужны. Дождитесь завершения каждого запуска перед изменением Variable.

6.1. Выбрать ветку через Variable

Основная практика. Сложность: средняя.

В Admin > Variables создайте ключ branch_format со значением csv. Запустите DAG, затем смените значение на json и сделайте новый запуск. Посмотрите граф и лог merge_results.

Что должно получиться: при csv CSV-ветка получает success, JSON-ветка - skipped; при json наоборот. merge_results выполняется в обоих случаях и читает XCom только текущего запуска. Правило none_failed_min_one_success допускает пропущенную ветку, если другая успешна.

Для третьего запуска задайте xml. choose_format должен упасть с понятным сообщением, DAG Run - завершиться ошибкой. Верните branch_format=csv и проверьте успешный запуск.

6.A. Добавить ветку ручной проверки

По желанию. Сложность: высокая.

Добавьте задачу review_branch и значение review в выбор формата. Для неизвестного значения направляйте выполнение в эту же ветку. Возвращайте из нее строку review, включите ее в зависимости перед слиянием и в список task_ids при чтении XCom в merge_results.

Что должно получиться: при review или неизвестном формате выполняется только новая ветка; две остальные пропущены, слияние и весь запуск успешны. Проверьте, что csv и json продолжают работать. Верните значение csv.

6.B. Прочитать CSV и JSON

По желанию. Сложность: высокая. Третья ветка из 6.A не требуется.

Создайте два входных файла в data/input: branch_records.csv и branch_records.json. Оба должны содержать две записи с id 1 и 2 и суммами amount 10 и 20. В функциях веток прочитайте соответствующий файл, проверьте положительность сумм и верните словарь с количеством записей и общей суммой. Слияние уже читает XCom выбранной ветки.

Что должно получиться: отдельные запуски для CSV и JSON дают по 2 записи и сумму 30. Сравните их логи вручную. В каждом запуске работает одна ветка; файл результата другой ветки из старого запуска для сравнения не нужен.

7. Ошибки и повторы

Используйте исходный error_handling_dag.py. Режим задается Variable error_mode; по умолчанию это retry_once. Обе рабочие задачи имеют одну повторную попытку с задержкой 10 секунд. Каждый сценарий начинайте новым запуском и ждите его завершения перед сменой режима.

7.1. Увидеть повторную попытку

Основная практика. Сложность: начальная.

Установите error_mode=retry_once и запустите DAG. Откройте retry_task и логи ее попыток. Первая намеренно падает, вторая завершается успешно. Состояние up_for_retry видно во время ожидания; если оно уже сменилось, причину и номера попыток можно найти в логах.

Что должно получиться: unreliable_task, retry_task и success_handler успешны, failure_handler пропущен, весь DAG Run успешен. В логе первой попытки есть временная учебная ошибка, во второй - сообщение о выполнении. retries=1 означает две попытки всего.

7.2. Проверить окончательную ошибку

Основная практика. Сложность: средняя.

Добавьте retries=0 непосредственно в оператор retry_task, дождитесь обновления кода и создайте новый запуск в режиме retry_once.

Что должно получиться: retry_task получает failed, success_handler - upstream_failed, а failure_handler выполняется. Весь DAG Run остается failed, хотя обработчик ошибки выполнился успешно.

Уберите локальное переопределение retries. Теперь по очереди запустите режимы fail_first и fail_second. Обработчик должен заметить отказ любой из двух задач. В графе проверьте, что каждый обработчик зависит от обеих рабочих задач. Верните error_mode=retry_once и получите успешный запуск.

7.3. Добавить свое сообщение в лог

Основная практика. Сложность: начальная.

В handle_success() добавьте logging.info() с собственным сообщением о завершении. У logging есть уровни INFO, WARNING и ERROR; print тоже попадает в лог, но не позволяет явно выбрать уровень сообщения.

Что должно получиться: после успешного запуска сообщение есть в логе success_handler. При окончательном отказе эта функция не выполняется, поэтому сообщения о всеобщем успехе нет.

7.A. Сравнить задержки

По желанию. Сложность: высокая.

В режиме fail_second задайте для retry_task три повтора, retry_delay=timedelta(seconds=5) и retry_exponential_backoff=True. Сопоставьте время окончания неудачной попытки с началом следующей. Задержка увеличивается; планировщик может добавить время ожидания, поэтому точного удвоения по часам требовать не нужно.

Что должно получиться: четыре неуспешные попытки, после них failure_handler и итог failed. Верните параметры повторов к исходным и режим к retry_once.

7.B. Добавить callback

По желанию. Сложность: высокая.

Добавьте функцию notify_failure(context) и укажите ее в on_failure_callback рабочих задач. В ней используйте logging.error и выведите context['ti'].task_id и context['exception']. В разных опытах бросьте ConnectionError, ValueError и OSError с различимыми сообщениями. Настройка настоящей почты для этого задания не нужна.

Что должно получиться: после исчерпания повторов callback сообщает, какая задача и почему упала. При временной ошибке с последующим успехом он не вызывается. В этом примере callback назначен оператору: откройте Logs упавшей задачи и выберите последнюю попытку. После сообщения об отказе найдите свой текст из callback.

Если нужно наблюдать каждый повтор, для этого есть отдельный on_retry_callback. Верните исходные исключения и error_mode=retry_once перед переходом дальше.

8. Пулы, XCom и TaskGroup

Используйте исходный resource_management_dag.py. Он имитирует чтение двух источников по 20 секунд и передает небольшие числа через XCom. Ни таблицы, ни файлы из предыдущих заданий ему не нужны.

8.1. Ограничить доступные слоты

Основная практика. Сложность: средняя.

Перед запуском в Admin > Pools создайте training_pool с двумя слотами. В DAG задача read_customers занимает два слота, read_orders - один. Запустите DAG и наблюдайте за задачами и пулом.

Что должно получиться: эти две задачи выполняются последовательно, хотя между ними нет зависимости: вместе им нужны три слота. Во время выполнения одной другая ожидает. Если пул не создан, задачи тоже будут ждать, поэтому сначала проверьте его имя и размер.

Поменяйте pool_slots у read_customers на 1 и повторите запуск. Теперь обе задачи могут работать одновременно, если свободны ресурсы executor. Сравните интервалы выполнения, затем верните значение 2.

8.2. Добавить метрику в XCom

Основная практика. Сложность: средняя.

В UI откройте XCom задачи calculate_metrics: исходный словарь содержит customers=20 и orders=30. Добавьте в возвращаемый словарь total_rows, равный сумме этих двух значений. Запустите DAG и проверьте XCom и лог log_metrics.

Что должно получиться: в обоих местах total_rows=50. Возвращаемое Python-функцией значение сохранено под ключом return_value и доступно следующей задаче через xcom_pull.

8.3. Сгруппировать задачи

Основная практика. Сложность: средняя.

Импортируйте TaskGroup из airflow.utils.task_group. Внутри блока with DAG(...) поместите определения двух задач чтения в with TaskGroup("extract"):. Метрики и лог оставьте снаружи, зависимости между задачами сохраните.

После группировки полные идентификаторы станут extract.read_customers и extract.read_orders. Обновите обе ссылки в calculate_metrics().

Что должно получиться: в графе появилась сворачиваемая группа. Нажмите на ее название со стрелкой, чтобы увидеть две задачи внутри. Метрика по-прежнему равна 50; ограничения пула продолжают действовать.

8.A. Подставить XCom в шаблон

По желанию. Сложность: средняя. Требуется 8.2; группировку 8.3 можно сохранить.

Добавьте после calculate_metrics задачу с BashOperator. В bash_command подставьте {{ ti.xcom_pull(task_ids='calculate_metrics')['total_rows'] }} в команду echo. Назовите задачу print_total.

Что должно получиться: в Details задачи print_total, в разделе Rendered Templates, уже подставлено число 50. В логе команды тоже 50.

8.B. Отправить уведомление

По желанию. Сложность: высокая. Требуется 7.B: вернитесь к error_handling_dag.py с callback отказа. Нужен отдельно настроенный SMTP и адрес, на который вы можете отправить тестовое письмо. В стандартном стенде SMTP не настроен; отсутствие письма здесь не означает ошибку основного маршрута.

Используйте пример из раздела об алертинге. Добавьте отправку письма в callback отказа. Укажите DAG, задачу, идентификатор запуска и причину ошибки. Для успешного DAG Run задайте отдельный on_success_callback у DAG(...).

Что должно получиться: после постоянной ошибки приходит сообщение об отказе, после успешного запуска - об успехе. Уведомление не меняет результат обработки данных. Запуск с отказом сохраняет failed; ошибку доставки ищите в логах callback. Возвращение к основной практике не требует настройки SMTP: отключите почтовый callback и верните исходный режим ошибок.

Когда можно двигаться дальше

Основная практика закончена, когда вы можете найти свой запуск и его логи, добавить задачу по образцу, объяснить выбранную ветку и повторную попытку, проверить файл или строки в таблице и прочитать XCom. По последнему блоку достаточно показать ожидание слотов и работающую TaskGroup.

Задания по желанию расширяют эти опыты. Их выполнение не требуется для перехода к следующему блоку программы.