441 lines
46 KiB
Markdown
441 lines
46 KiB
Markdown
# Практика по Apache Airflow
|
||
|
||
Здесь мы знакомимся с Airflow на готовых примерах: запускаем DAG, меняем задачи и смотрим, что произошло. Свой цельный процесс обработки данных вы соберете в курсовых работах дальше по программе.
|
||
|
||
## Как проходить практику
|
||
|
||
В каждом блоке сначала выполните основную практику. Задания "По желанию" находятся в конце блока: их можно пропустить и перейти дальше. Сложность указана отдельно. Если дополнительному заданию нужны результаты другого упражнения, это написано в его начале.
|
||
|
||
Основные задания одного блока выполняйте последовательно, сохраняя изменения. Перед новым блоком прочитайте условия начала. Загрузка CSV и проверки DQ используют одну таблицу, поэтому идут вместе.
|
||
|
||
| Блок | Примеры | Что наблюдаем | Перед практикой |
|
||
|---|---|---|---|
|
||
| [1. Первый DAG](#first-dag) | `hello_world_dag.py` | Задачи, зависимости, операторы, расписание | Главы 4 и 5; для 1.3 - [расписания][schedules] |
|
||
| [2. SQL и подключения](#sql) | `sql_basic_dag.py` | Connection, SQL, ошибка подключения | [Подключения][connections] |
|
||
| [3. Работа с файлами](#files) | `file_operations_dag.py` | Результаты отдельных задач | Главы 5 и 6 |
|
||
| [4. Загрузка и DQ](#load-dq) | `csv_to_postgres.py`, `csv_to_postgres_dq.py` | XCom, данные в БД, ошибка проверки | Глава 6, [шаблоны][templates] и [XCom][xcom] |
|
||
| [5. Разбор ETL](#etl) | `data_processing_dag.py` | Параллельные шаги и путь данных | Глава 5 |
|
||
| [6. Ветвление](#branching) | `branching_dag.py` | Variables, выбранная ветка, `skipped` | Глава 6 и [Variables][variables] |
|
||
| [7. Ошибки и повторы](#errors) | `error_handling_dag.py` | Попытки, обработчики, итог запуска | Глава 6; для 7.B - [callback][alerts] |
|
||
| [8. Пулы и группировка](#resources) | `resource_management_dag.py` | Слоты, XCom, TaskGroup | [Пулы][pools], [XCom][xcom] и [TaskGroup][taskgroups] |
|
||
|
||
[schedules]: 08%20-%20Управление%20временем.md#настройка-автоматических-запусков
|
||
[connections]: 07%20-%20Шаблоны,%20переменные%20и%20подключения%20в%20Airflow.md#централизованные-подключения-connections
|
||
[templates]: 07%20-%20Шаблоны,%20переменные%20и%20подключения%20в%20Airflow.md#динамические-шаблоны-airflow-на-основе-jinja
|
||
[variables]: 07%20-%20Шаблоны,%20переменные%20и%20подключения%20в%20Airflow.md#безопасные-переменные-variables
|
||
[xcom]: 09%20-%20Пулы,%20XCom,%20TaskGroup%20и%20алертинг%20в%20Airflow.md#обмен-данными-между-задачами-xcom-и-контекст-выполнения
|
||
[pools]: 09%20-%20Пулы,%20XCom,%20TaskGroup%20и%20алертинг%20в%20Airflow.md#управление-ресурсами-с-помощью-пулов-задач
|
||
[taskgroups]: 09%20-%20Пулы,%20XCom,%20TaskGroup%20и%20алертинг%20в%20Airflow.md#группировка-задач-с-помощью-taskgroup
|
||
[alerts]: 09%20-%20Пулы,%20XCom,%20TaskGroup%20и%20алертинг%20в%20Airflow.md#система-оповещений-алертинг
|
||
|
||
Главы доступны в [оглавлении учебника](README.md). Инструкция запуска и параметры учебной БД находятся в [README стенда](airflow-docker/README.md), файлы примеров - в [airflow-docker/dags/](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 можно открыть консоль учебной БД:
|
||
|
||
```bash
|
||
docker compose exec postgres-training psql -U student -d training
|
||
```
|
||
|
||
В ней выполняются SQL-запросы из заданий; выход - `\q`. Сохраняйте исходные DAG перед опытами вне каталога `dags`: две Python-копии с одинаковым `dag_id` мешают Airflow выбрать нужную.
|
||
|
||
<a id="first-dag"></a>
|
||
|
||
## 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`.
|
||
|
||
После запуска проверьте файл:
|
||
|
||
```bash
|
||
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; ограничение можно оставить.
|
||
|
||
<a id="sql"></a>
|
||
|
||
## 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.
|
||
|
||
Проверьте таблицу:
|
||
|
||
```sql
|
||
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`. Для уже созданной таблицы сначала выполните:
|
||
|
||
```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`.
|
||
|
||
<a id="files"></a>
|
||
|
||
## 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`. В графе три задачи имеют общего предшественника, но не зависят друг от друга. Повтор не добавляет дубликаты: файл отдела перезаписывается.
|
||
|
||
<a id="load-dq"></a>
|
||
|
||
## 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.
|
||
|
||
Проверьте результат в БД:
|
||
|
||
```sql
|
||
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 успешны, сводка перечисляет результаты проверок.
|
||
|
||
Для отрицательного сценария поменяйте знак суммы одного заказа:
|
||
|
||
```sql
|
||
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'`. Существующую таблицу подготовьте один раз:
|
||
|
||
```sql
|
||
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-запуск.
|
||
|
||
<a id="etl"></a>
|
||
|
||
## 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 этой проверки находится в логе ее последней попытки.
|
||
|
||
<a id="branching"></a>
|
||
|
||
## 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. Сравните их логи вручную. В каждом запуске работает одна ветка; файл результата другой ветки из старого запуска для сравнения не нужен.
|
||
|
||
<a id="errors"></a>
|
||
|
||
## 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` перед переходом дальше.
|
||
|
||
<a id="resources"></a>
|
||
|
||
## 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 не настроен; отсутствие письма здесь не означает ошибку основного маршрута.
|
||
|
||
Используйте пример из [раздела об алертинге](09%20-%20Пулы,%20XCom,%20TaskGroup%20и%20алертинг%20в%20Airflow.md#система-оповещений-алертинг). Добавьте отправку письма в callback отказа. Укажите DAG, задачу, идентификатор запуска и причину ошибки. Для успешного DAG Run задайте отдельный `on_success_callback` у `DAG(...)`.
|
||
|
||
**Что должно получиться:** после постоянной ошибки приходит сообщение об отказе, после успешного запуска - об успехе. Уведомление не меняет результат обработки данных. Запуск с отказом сохраняет `failed`; ошибку доставки ищите в логах callback. Возвращение к основной практике не требует настройки SMTP: отключите почтовый callback и верните исходный режим ошибок.
|
||
|
||
## Когда можно двигаться дальше
|
||
|
||
Основная практика закончена, когда вы можете найти свой запуск и его логи, добавить задачу по образцу, объяснить выбранную ветку и повторную попытку, проверить файл или строки в таблице и прочитать XCom. По последнему блоку достаточно показать ожидание слотов и работающую TaskGroup.
|
||
|
||
Задания по желанию расширяют эти опыты. Их выполнение не требуется для перехода к следующему блоку программы.
|