diff --git a/09 - Пулы, XCom, TaskGroup и алертинг в Airflow.md b/09 - Пулы, XCom, TaskGroup и алертинг в Airflow.md
index 8e1443b..f197074 100644
--- a/09 - Пулы, XCom, TaskGroup и алертинг в Airflow.md
+++ b/09 - Пулы, XCom, TaskGroup и алертинг в Airflow.md
@@ -8,9 +8,9 @@
В системах с высокой нагрузкой, где одновременно запускается множество задач и DAG-ов, может возникнуть чрезмерная нагрузка на исполнителей и серверную часть. Это может привести к ошибкам выполнения и даже к отказу системы, если не установить соответствующие ограничения.
-В Airflow для решения этой проблемы существует механизм управления ресурсами — **пулы задач** (pools). По умолчанию в Airflow настроен один пул задач — `default_pool` с 128 слотами, что означает возможность параллельного выполнения 128 задач одновременно. Пул `default_pool` нельзя удалить, но можно изменить его размер — увеличить или уменьшить количество слотов.
+В Airflow для решения этой проблемы существует механизм управления ресурсами — **пулы задач** (pools). По умолчанию в Airflow настроен один пул задач — `default_pool` с 128 слотами. При стандартном `pool_slots=1` это верхняя граница в 128 одновременно работающих задач; ограничения executor и DAG могут уменьшить ее. Пул `default_pool` нельзя удалить, но можно изменить его размер — увеличить или уменьшить количество слотов.
-Когда планировщик обнаруживает, что наступило время выполнения DAG, он запускает задачу согласно заданной последовательности. При этом задача занимает один слот в пуле и освобождает его после завершения.
+Когда планировщик обнаруживает, что наступило время выполнения DAG, он запускает задачу согласно заданной последовательности. По умолчанию задача занимает один слот в пуле и освобождает его после завершения.
Создать новый пул задач и установить его размер можно через веб-интерфейс Airflow. Рассмотрим пример пула `data_processing_pool` для тяжёлых задач:
@@ -29,7 +29,7 @@
- Организовать запуск процессов в системе
- Предотвратить перегрузку системы при выполнении большого количества ресурсоемких задач
-Вы можете задать "вес" задачи через параметр `pool_slots`, чтобы оптимизировать распределение нагрузки. Если общее количество задач превышает доступные слоты, планировщик поставит задачу в очередь и запустит ее, как только появятся свободные ресурсы.
+Вы можете задать "вес" задачи через параметр `pool_slots`, чтобы оптимизировать распределение нагрузки. Планировщик учитывает сумму занятых слотов. Например, в пуле из двух слотов задачи с `pool_slots=2` и `pool_slots=1` не смогут работать одновременно: одна ждет завершения другой.
Пример настройки веса задач:
@@ -58,7 +58,7 @@ BashOperator(
)
```
-Более подробную информацию о механизме пулов можно найти в [официальной документации](https://airflow.apache.org/docs/apache-airflow/stable/concepts/pools.html#pools).
+Более подробную информацию о механизме пулов можно найти в [официальной документации](https://airflow.apache.org/docs/apache-airflow/2.9.2/administration-and-deployment/pools.html).
## Обмен данными между задачами: XCom и контекст выполнения
@@ -220,100 +220,64 @@ TaskGroup — это удобный способ логической групп
## Система оповещений (алертинг)
-Алертинг — один из ключевых компонентов системы оркестрации, так как важно своевременно получать уведомления об ошибках для их оперативного анализа и решения.
-
-В Airflow есть встроенная поддержка отправки уведомлений на электронную почту (при условии, что в конфигурации настроен SMTP-сервер). При создании DAG указываются email-адреса, на которые будут отправляться сообщения. С помощью параметров можно настроить различные сценарии оповещений.
-
-Давайте модифицируем наш первый DAG так, чтобы получать уведомления на почту при возникновении ошибок. При этом настроим перезапуск задач в случае неудачи (например, 2 попытки), но без уведомлений о самих перезапусках:
-
-В приведенном примере создан DAG с идентификатором 'customer_analysis_pipeline', который настроен на отправку уведомлений по электронной почте только при ошибках (email_on_failure=True), но не при повторных попытках (email_on_retry=False). Также установлено 2 попытки повторного запуска задач при ошибках с задержкой 2 минуты между попытками.
+Для первого опыта достаточно callback с записью в лог. Он получает контекст задачи и помогает связать сообщение с конкретным запуском:
```python
-import os
-import datetime as dt
-import pandas as pd
-from airflow.models import DAG
-from airflow.operators.python import PythonOperator
-from airflow.operators.bash import BashOperator
-from sqlalchemy import create_engine
+import logging
-# основные параметры DAG
-args = {
- 'owner': 'data_engineering_team',
- 'start_date': dt.datetime(2021, 6, 15),
- 'retries': 2,
- 'retry_delay': dt.timedelta(minutes=2),
- 'email': ["data-team@example.com"],
- 'email_on_failure': True,
- 'email_on_retry': False,
-}
-dag = DAG(
- dag_id='customer_analysis_pipeline',
- schedule_interval=None,
- default_args=args,
-)
+def notify_failure(context):
+ ti = context["ti"]
+ logging.error(
+ "Ошибка: dag=%s task=%s run=%s причина=%s",
+ ti.dag_id, ti.task_id, context["run_id"], context["exception"],
+ )
```
-Теперь вы будете получать email-уведомления при ошибках выполнения.
+Укажите `on_failure_callback=notify_failure` в рабочем операторе. Функция вызывается после окончательного отказа, когда повторы исчерпаны. Для сообщения о предстоящем повторе существует `on_retry_callback`. Например, при `retries=2` задача может выполниться три раза: исходная попытка и два повтора.
-Пример функции для отправки уведомлений с использованием параметров из контекста:
+Callback запускается при реальном выполнении задачи. Ручная смена статуса в UI его не проверяет. Смотрите лог самой задачи и логи scheduler; ошибки callback ищите в логах scheduler. [Документация Airflow 2.9.2 о callbacks](https://airflow.apache.org/docs/apache-airflow/2.9.2/administration-and-deployment/logging-monitoring/callbacks.html).
+
+### Отправка почты по желанию
+
+В учебном стенде SMTP не настроен. Для отправки нужен SMTP-сервер и адрес получателя, которому можно отправлять тестовые сообщения. Укажите настройки в общем окружении сервисов Airflow: `AIRFLOW__SMTP__SMTP_HOST`, `AIRFLOW__SMTP__SMTP_PORT`, `AIRFLOW__SMTP__SMTP_MAIL_FROM`, параметры TLS/SSL и учетные данные вашего сервера. Секреты храните локально, вне Git. После изменения окружения пересоздайте сервисы командой `docker compose up -d`. Полный список параметров находится в [конфигурации Airflow 2.9.2](https://airflow.apache.org/docs/apache-airflow/2.9.2/configurations-ref.html#smtp).
+
+После настройки замените запись в лог отправкой сообщения:
```python
-from datetime import datetime, timedelta, timezone
-import dateutil
+from html import escape
+
from airflow.utils.email import send_email_smtp
-from airflow.operators.python import get_current_context
-MAIL_LIST = [
- "email_1@gmail.ru",
- "email_2@gmail.ru"
-]
+MAIL_TO = ["student@example.com"] # Замените своим тестовым адресом.
-def notify_email(calculation_dt: str, dagrun_begin_time: str):
- # dag_run date is utc timezone, so add `timezone.utc` to calculation_end to combat 3 hour diff.
- context = get_current_context()
- calculation_start = dateutil.parser.isoparse(dagrun_begin_time.replace('Z', '+00:00'))
- calculation_end = datetime.now(timezone.utc)
- duration = str(timedelta(seconds=(calculation_end - calculation_start).seconds))
-
- title = f"Ежедневные расчёты завершились успешно (dag: {context['task_instance'].dag_id})."
-
- body = f"""
- Привет,
- Я закончил работу над расчётом за "{calculation_dt}". Это заняло {duration} часов/минут/секунд.
- Логи и запуски тоже можно посмотреть тут.
-
-
-
-
- Навеки твой,
- Airflow бот
- """
-
- send_email_smtp(";".join(MAIL_LIST), title, body)
+
+def notify_failure(context):
+ ti = context["ti"]
+ send_email_smtp(
+ to=MAIL_TO,
+ subject=f"Ошибка {ti.dag_id}: {ti.task_id}",
+ html_content=(
+ f"Запуск: {escape(context['run_id'])}
"
+ f"Причина: {escape(str(context['exception']))}
"
+ f'Лог задачи
'
+ ),
+ )
+
+
+def notify_success(context):
+ send_email_smtp(
+ to=MAIL_TO,
+ subject=f"DAG {context['dag'].dag_id} завершен успешно",
+ html_content=f"Запуск: {escape(context['run_id'])}
",
+ )
```
-Такую функцию можно разместить в отдельном файле (например, `utils.py`), импортировать как модуль в нужных DAG и вызывать отдельной задачей:
+Для примера с ошибками задайте `on_failure_callback=notify_failure` у двух рабочих задач, а `on_success_callback=notify_success` у самого `DAG(...)`. Если упадут обе задачи, придут два сообщения об отказе. Сообщение об успехе отправится только после успешного DAG Run. Не включайте одновременно `email_on_failure=True`, если не хотите получать еще и встроенное письмо о той же ошибке.
-```python
-from airflow.operators.python import PythonOperator
-from airflow.utils.trigger_rule import TriggerRule
-from utils import notify_email
+Не добавляйте ради уведомления единственную завершающую задачу с `trigger_rule="all_done"`: если она успешна, Airflow может признать весь запуск успешным, несмотря на отказ внутри графа. Callback не меняет граф и итог обработки. Ошибка отправки означает проблему уведомления; она не превращает успешно обработанные данные в неуспешные и не скрывает исходный отказ. Подробности определения результата - в [описании DAG Run](https://airflow.apache.org/docs/apache-airflow/2.9.2/core-concepts/dag-run.html#dag-run-status).
-LOCAL_CALCULATION_DT = '{{ dag.timezone.convert(execution_date).strftime("%Y-%m-%d") }}'
-DAG_RUN_BEGIN_TIME = "{{ dag_run.start_date }}"
-
-email_notification_task = PythonOperator(
- task_id="send_email_notification",
- python_callable=notify_email,
- dag=dag,
- trigger_rule=TriggerRule.ALL_DONE,
- op_args=[LOCAL_CALCULATION_DT, DAG_RUN_BEGIN_TIME],
-)
-
-... >> email_notification_task
-```
+Проверка: сначала воспроизведите окончательный отказ, затем успешный запуск. Сверьте текст каждого письма с состоянием соответствующего DAG Run. Упражнения находятся в [блоке ошибок](airflow-docker/educational-tasks.md#errors) и [задании на почту](airflow-docker/educational-tasks.md#resources).
## Практическое применение: улучшенный DAG
diff --git a/README.md b/README.md
index 66660a9..3c7f871 100644
--- a/README.md
+++ b/README.md
@@ -1,19 +1,8 @@
# Учебник по Apache Airflow для начинающих
-Добро пожаловать в учебник по Apache Airflow! Этот курс создан специально для тех, кто уже освоил Python и SQL и готов познакомиться со своим первым инструментом для ETL-процессов. К учебнику прилагается готовый практический стенд, позволяющий сразу применить полученные знания на практике.
+Учебник знакомит с Apache Airflow 2.9: устройством DAG, запуском задач, интерфейсом и поиском ошибок. К нему прилагается Docker-стенд с готовыми примерами. Перед началом нужны Python, pandas, SQL и основы Docker.
-## 📚 Введение
-
-Apache Airflow — это мощный оркестратор рабочих процессов с открытым исходным кодом, который стал стандартом де-факто в мире обработки данных. Если вы работаете с автоматизацией процессов обработки информации — будь то подготовка данных для машинного обучения, создание отчетов или интеграция данных из различных источников — Airflow поможет вам организовать эти процессы эффективно и надежно.
-
-Этот учебник разработан с учетом того, что вы:
-- Имеете базовые знания Python и SQL
-- Хотите освоить свой первый ETL-инструмент
-- Нуждаетесь в практическом руководстве с понятными объяснениями
-
-Мы начнем с основ и постепенно перейдем к продвинутым возможностям, чтобы вы могли уверенно использовать Airflow в реальных проектах. Курс ориентирован на версию Airflow 2.9 и использует практический подход с множеством примеров и визуальных материалов.
-
-## 🎯 Что вы узнаете
+## Что вы узнаете
- **Основы Airflow**: что это такое, для каких задач используется и почему он стал таким популярным
- **Ключевые концепции**: DAG, задачи, операторы и другие фундаментальные понятия
@@ -23,7 +12,7 @@ Apache Airflow — это мощный оркестратор рабочих п
- **Отладка и мониторинг**: как понимать статусы задач и быстро находить проблемы
- **Продвинутые возможности**: шаблоны, параметры, управление временем, пулы, XCom, TaskGroup и алертинг
-## 📖 Содержание курса
+## Содержание курса
### [01. Введение в Airflow](01%20-%20Введение%20в%20Airflow.md)
- Почему Apache Airflow стал незаменимым инструментом для работы с данными
@@ -73,34 +62,25 @@ Apache Airflow — это мощный оркестратор рабочих п
- Логическая группировка задач с TaskGroup
- Настройка системы оповещений и уведомлений
-## 🛠️ Практический стенд
+## Практический стенд
-К этому учебнику прилагается полноценный стенд Apache Airflow, который позволяет сразу приступить к практике и закрепить полученные знания. В папке [`airflow-docker`](airflow-docker/) вы найдете готовую среду для экспериментов с 6 готовыми DAG'ами разной сложности и практическими заданиями для каждого из них.
+В [`airflow-docker`](airflow-docker/) находятся Airflow 2.9.2, учебная PostgreSQL и девять готовых DAG. [Инструкция запуска](airflow-docker/README.md) поможет поднять стенд, а [маршрут практики](airflow-docker/educational-tasks.md) связывает восемь блоков заданий с главами учебника.
-Стенд включает настроенную среду с PostgreSQL для тренировок, образцы данных и детальные инструкции для практических занятий. Подробную информацию о возможностях стенда и инструкции по запуску смотрите в [`README.md`](airflow-docker/README.md) папки `airflow-docker`.
+Здесь вы запускаете готовый DAG, меняете одну часть и проверяете результат в графе, логах или данных. Самостоятельные процессы обработки данных относятся к курсовым работам дальше по [программе](https://de.dementev.space/).
-## 🚀 Как использовать этот учебник
+## Как использовать этот учебник
-1. **Изучайте теорию**: начните с Введения в Airflow и переходите к следующим главам по порядку
-2. **Практикуйтесь на стенде**: после каждой главы выполняйте соответствующие задания в практическом стенде
-3. **Создавайте свои DAG**: применяйте полученные знания для решения реальных задач
-4. **Используйте визуальные материалы**: в учебнике много изображений и диаграмм для лучшего понимания
-5. **Экспериментируйте**: не бойтесь пробовать разные настройки и сценарии на учебном стенде
+1. Прочитайте главы 1-5 и запустите стенд. Перед каждым блоком практики сверяйтесь с колонкой "Перед практикой": некоторые задания используют материал следующих глав.
+2. Выполняйте основные задания блока по порядку и сохраняйте изменения. Каждый новый блок описывает, с какого состояния начать.
+3. Проверяйте раздел "Что должно получиться". В опыте с ошибкой результатом может быть `failed`; важно найти причину и восстановить успешный запуск.
+4. Задания "По желанию" можно пропустить. Они не нужны для продолжения основного маршрута, а высокая сложность отмечена отдельно.
-## 📝 Технические требования
+Практика завершена, когда пройдены основные задания всех восьми блоков и вы можете объяснить наблюдаемое поведение. Выполнение всех усложнений для этого не требуется.
-- **Python**: базовое знание языка
+## Технические требования
+
+- **Python и pandas**: чтение функций, работа с таблицами
- **SQL**: понимание основных запросов и концепций
- **Airflow**: версия 2.9 (на которой ориентирован курс)
+- **Docker и Docker Compose**: запуск учебного стенда
- **Среда разработки**: любой текстовый редактор или IDE с поддержкой Python
-
-## 💡 Советы для успешного обучения
-
-- **Не торопитесь**: лучше глубоко понять каждый концепт, чем быстро пройти весь курс
-- **Задавайте вопросы**: если что-то непонятно, вернитесь к предыдущим материалам
-- **Делайте заметки**: записывайте важные моменты и примеры кода
-- **Применяйте на практике**: сразу пробуйте создавать свои DAG для реальных задач
-
----
-
-*Этот учебник создан для начинающих специалистов в области данных и инженерии. Все материалы ориентированы на практическое применение и пошаговое освоение Apache Airflow.*
diff --git a/airflow-docker/README.md b/airflow-docker/README.md
index 444fc85..e616b9b 100644
--- a/airflow-docker/README.md
+++ b/airflow-docker/README.md
@@ -1,277 +1,106 @@
-# Educational Airflow Setup for Beginners
+# Учебный стенд Apache Airflow
-Простой учебный стенд Apache Airflow для начинающих, изучающих SQL и Python.
+Стенд содержит Airflow 2.9.2, учебную PostgreSQL и девять DAG. Он нужен для знакомства с инструментом: запустить пример, изменить его и посмотреть результат. Для работы понадобятся Docker с Compose, браузер и знания Python, pandas и SQL.
-## 🎯 Цель проекта
+## Запуск
-Создать максимально простую среду для изучения Apache Airflow.
-Для удобства все переменные окружения захардкожены в docker скрипты. На проде так делать не надо :)
-
-## 📋 Предварительные требования
-
-- Docker и Docker Compose
-- Базовые знания Python и SQL
-- Веб-браузер для доступа к интерфейсу Airflow
-
-## 🚀 Быстрый старт
-
-### 1. Запуск стенда
+Из корня репозитория:
```bash
-# Перейдите в директорию проекта
cd airflow-docker
-
-# Запустите все сервисы
-docker-compose up -d
+docker compose up -d --build
```
-### 2. Доступ к интерфейсам
+Команды подходят для Linux и PowerShell в Windows. При первом запуске Docker собирает образ, поднимает две БД, выполняет `airflow-init`, затем запускает webserver и scheduler. Инициализация создает учетную запись и подключение `postgres_training`.
-- **Airflow UI**: http://localhost:8080
- - Логин: `admin`
- - Пароль: `admin`
+Проверьте состояние:
-- **PostgreSQL для тренировок**: `localhost:5432`
- - База данных: `training`
- - Пользователь: `student`
- - Пароль: `student`
-
-- **PostgreSQL для метаданных Airflow**: `localhost:5434`
- - База данных: `airflow`
- - Пользователь: `airflow`
- - Пароль: `airflow`
-
-## 🏗️ Архитектура стенда
-
-```mermaid
-graph LR
-
- %% Локальная файловая система
- subgraph local["Локальная файловая система"]
- dags["Каталог
Educational DAGs"]
- fs["Local Filesystem"]
- dags --> fs
- end
-
- %% Docker-контейнеры
- subgraph docker["Контейнеры Docker"]
- subgraph airflow["Airflow"]
- ws["Airflow Webserver"]
- sch["Airflow Scheduler"]
- meta["PostgreSQL Metadata"]
- sch --> meta
- ws --> meta
- end
-
- subgraph training["Учебная БД"]
- train["PostgreSQL Training"]
- end
- end
-
- %% Клиенты
- subgraph clients["Клиенты"]
- user["Пользователь
(браузер)"]
- sql["SQL клиент"]
- end
-
- %% Монтирование DAG'ов в контейнеры Airflow
- dags -. "том с DAG-файлами" .- ws
- dags -. "том с DAG-файлами" .- sch
-
- %% Airflow-задачи работают с учебной БД
- sch -- "задачи DAG" --> train
-
- %% Внешние подключения
- user -- "порт 8080" --> ws
- sql -- "порт 5432" --> train
+```bash
+docker compose ps -a
+docker compose logs airflow-init
```
-## 📁 Структура проекта
+Завершение `airflow-init` с кодом 0 нормально: это разовая подготовка. Остальные четыре сервиса должны работать. Откройте http://localhost:8080 и войдите как `admin` / `admin`.
-```
+| Сервис | Адрес с компьютера | База | Пользователь / пароль |
+|---|---|---|---|
+| Airflow UI | `http://localhost:8080` | - | `admin` / `admin` |
+| Учебная PostgreSQL | `localhost:5432` | `training` | `student` / `student` |
+| Метаданные Airflow | `localhost:5434` | `airflow` | `airflow` / `airflow` |
+
+В задачах используется адрес `postgres-training:5432`: это имя сервиса внутри Docker. Базу метаданных для упражнений не используйте. Учебные пароли заданы в Compose; для публичного сервера такая конфигурация не подходит.
+
+## Прохождение практики
+
+Откройте [задания](educational-tasks.md). Их восемь блоков проходят последовательно; перед каждым указаны нужные главы [учебника](../README.md), исходные файлы и настройки.
+
+Основные задания внутри блока сохраняют изменения предыдущих шагов. Дополнительные находятся в конце блока и помечены "По желанию"; их можно пропустить. Для завершения практики достаточно основных заданий и проверки их результатов.
+
+| Блок | DAG |
+|---|---|
+| [Первый DAG](educational-tasks.md#first-dag) | `hello_world_dag` |
+| [SQL и подключения](educational-tasks.md#sql) | `sql_basic_dag` |
+| [Работа с файлами](educational-tasks.md#files) | `file_operations_dag` |
+| [Загрузка CSV и DQ](educational-tasks.md#load-dq) | `csv_to_postgres`, `csv_to_postgres_dq` |
+| [Разбор ETL](educational-tasks.md#etl) | `data_processing_dag` |
+| [Ветвление](educational-tasks.md#branching) | `branching_dag` |
+| [Ошибки и повторы](educational-tasks.md#errors) | `error_handling_dag` |
+| [Пулы, XCom и TaskGroup](educational-tasks.md#resources) | `resource_management_dag` |
+
+Описание поведения исходных примеров находится в [справочнике DAG](dag-specifications.md). Почта нужна только для отдельного задания по желанию; SMTP в стенде не настроен.
+
+## Файлы и данные
+
+```text
airflow-docker/
-├── docker-compose.yml # Конфигурация Docker
-├── .env # Файл не используется: переменные заданы в docker-compose.yml
-├── dags/ # DAG файлы для обучения
-│ ├── hello_world_dag.py # Базовый пример
-│ ├── sql_basic_dag.py # Работа с SQL
-│ ├── file_operations_dag.py # Обработка файлов
-│ ├── csv_to_postgres.py # Загрузка CSV в Postgres (ETL)
-│ ├── csv_to_postgres_dq.py # Проверки качества данных (DQ)
-│ ├── data_processing_dag.py # Сложный ETL пайплайн
-│ ├── branching_dag.py # Условная логика
-│ └── error_handling_dag.py # Обработка ошибок
-├── data/ # Данные и артефакты прогонов
-│ ├── input/ # Входные данные
-│ └── output/ # Сгенерированные CSV, отчеты и результаты обработки
-├── logs/ # Логи Airflow
-├── README.md # Эта инструкция
-└── educational-tasks.md # Практические задания для студентов
+├── docker-compose.yml # Сервисы, порты и учебные подключения
+├── Dockerfile # Airflow 2.9.2
+├── requirements.txt # Дополнительные Python-пакеты
+├── dags/ # Девять учебных DAG
+├── data/input/ # Входные CSV и данные упражнений
+├── data/output/ # Результаты задач
+├── init-postgres.sql # Начальная подготовка учебной БД
+└── educational-tasks.md # Маршрут и условия заданий
```
-## 🎓 Учебные материалы
+`dags/` и `data/` смонтированы в контейнеры как `/opt/airflow/dags/` и `/opt/airflow/data/`. Правки DAG подхватываются планировщиком; перед новым запуском дождитесь обновления вкладки Code. Резервные копии Python-файлов храните вне `dags/`, чтобы не создать два определения одного `dag_id`.
-### Основы Airflow
+Таблицы PostgreSQL сохраняются в именованных томах `pg_data` и `pgmeta`. Обычная остановка их не удаляет. Логи Airflow хранятся внутри контейнеров; важные логи опыта сохраните до удаления контейнеров.
-**Цели:**
-- Понимание структуры DAG
-- Создание простых задач
-- Настройка зависимостей между задачами
-
-**Примеры DAG:**
-- `hello_world_dag.py` - базовые операторы Python
-- `sql_basic_dag.py` - работа с базами данных
-
-### Интеграция с данными
-
-**Цели:**
-- Подключение к PostgreSQL
-- Выполнение SQL запросов
-- Обработка файлов CSV
-
-**Примеры DAG:**
-- `file_operations_dag.py` - работа с файлами
-- `csv_to_postgres.py` - загрузка данных из CSV в PostgreSQL
-- `csv_to_postgres_dq.py` - автоматизированные проверки качества (Data Quality)
-- `data_processing_dag.py` - ETL процессы
-
-### Продвинутые возможности
-
-**Цели:**
-- Условное выполнение задач
-- Обработка ошибок
-- Параметризация workflows
-
-**Примеры DAG:**
-- `branching_dag.py` - условная логика
-- `error_handling_dag.py` - обработка ошибок
-
-## 🎯 Практические задания
-
-Для закрепления знаний по каждому DAG мы подготовили практические задания разного уровня сложности:
-
-### 📋 Учебные задания
-
-В файле **[educational-tasks.md](educational-tasks.md)** вы найдете детальные задания для каждого DAG:
-
-- 🌱 **Начальный уровень**: Базовые задачи по созданию и настройке DAG'ов
-- 📚 **Средний уровень**: Работа с данными, файлами и SQL операциями
-- 🚀 **Продвинутый уровень**: ETL процессы, ветвление, обработка ошибок
-
-### 🎓 Рекомендуемый путь обучения
-
-1. **Начните с `hello_world_dag.py`** - освоите основы Airflow
-2. **Перейдите к `sql_basic_dag.py`** - изучите работу с базами данных
-3. **Попрактикуйтесь на `file_operations_dag.py`** - работа с файлами
-4. **Освойте ETL и DQ на `csv_to_postgres.py` и `csv_to_postgres_dq.py`** - загрузка и валидация данных
-5. **Разберите сложный ETL на `data_processing_dag.py`** - обработка данных
-6. **Изучите продвинутые темы** - ветвление и обработка ошибок
-
-Каждое задание содержит:
-- Цель и сложность выполнения
-- Конкретные требования к модификации кода
-- Ожидаемый результат
-- Время на выполнение
-
-## 🔧 Технические детали
-
-### Переменные окружения
-
-В этом проекте мы не используем `.env`: все значения заданы напрямую в `docker-compose.yml`.
-
-Сервисы Airflow получают:
-- `POSTGRES_TRAINING_HOST=postgres-training`
-- `POSTGRES_TRAINING_PORT=5432`
-- `POSTGRES_TRAINING_DB=training`
-- `POSTGRES_TRAINING_USER=student`
-- `POSTGRES_TRAINING_PASSWORD=student`
-- `AIRFLOW_CONN_POSTGRES_TRAINING=postgresql://student:student@postgres-training:5432/training`
-
-`docker-compose run --rm airflow-init` запускает `airflow connections add postgres_training`, поэтому соединение доступно сразу после инициализации. Проверить наличие можно через:
+Для SQL-запросов из заданий:
```bash
-docker-compose exec airflow-webserver airflow connections get postgres_training
+docker compose exec postgres-training psql -U student -d training
```
-`csv_to_postgres.py` по умолчанию складывает сгенерированные CSV в `/opt/airflow/data/output`, то есть в локальный каталог `airflow-docker/data/output/`.
+Выход из psql - `\q`. Готовое подключение Airflow можно проверить командой:
-
-### Порты
-
-- `8080` - Airflow Webserver
-- `5432` - PostgreSQL для тренировок
-- `5434` - PostgreSQL для метаданных Airflow
-
-## 🛠️ Управление стендом
-
-### Запуск сервисов
```bash
-docker-compose up -d
+docker compose exec airflow-webserver airflow connections get postgres_training
```
-### Остановка сервисов
+## Управление и поиск ошибок
+
+Все команды ниже выполняются из `airflow-docker`.
+
```bash
-docker-compose down
+docker compose exec airflow-webserver airflow version
+docker compose exec airflow-webserver airflow dags list
+docker compose exec airflow-webserver airflow dags list-import-errors
+docker compose logs airflow-scheduler
+docker compose logs airflow-webserver
```
-### Просмотр логов
+Если DAG не появился, проверьте путь файла и ошибки импорта. Если задача упала, сначала откройте ее лог в UI. При ошибке подключения проверьте `docker compose ps` и Conn Id задачи. Задачи из блока 8 ждут, пока вы создадите указанный в задании пул.
+
+Остановка с сохранением данных:
+
```bash
-# Логи Airflow
-docker-compose logs airflow-webserver
-docker-compose logs airflow-scheduler
-
-# Логи PostgreSQL
-docker-compose logs postgres-training
-docker-compose logs postgres-metadata
+docker compose down
```
-### Перезапуск конкретного сервиса
-```bash
-docker-compose restart airflow-webserver
-```
-
-## 🐛 Решение проблем
-
-### DAG не появляется в интерфейсе
-- Проверьте, что файл находится в папке `dags/`
-- Убедитесь в правильности синтаксиса Python
-- Проверьте логи планировщика: `docker-compose logs airflow-scheduler`
-
-### Ошибки подключения к базе данных
-- Убедитесь, что PostgreSQL запущен: `docker-compose ps`
-- Проверьте логи PostgreSQL: `docker-compose logs postgres-training`
-
-### Задачи завершаются с ошибкой
-- Проверьте логи задачи в интерфейсе Airflow
-- Убедитесь в наличии необходимых Python пакетов
-
-## 📚 Дополнительные ресурсы
-
-- [Официальная документация Airflow](https://airflow.apache.org/docs/)
-- [Учебные материалы в родительской папке](../)
-- [Практические задания для студентов](educational-tasks.md)
-
----
-
-## 🌟 Начало работы с заданиями
-
-После запуска стенда Airflow, откройте файл [educational-tasks.md](educational-tasks.md) и выберите задание, соответствующее вашему уровню подготовки. Каждое задание содержит подробные инструкции и ожидаемые результаты.
-
-**Быстрый старт с заданиями:**
-```bash
-# Запустите стенд
-docker-compose up -d
-
-# Откройте задания в браузере или редакторе
-# И следуйте инструкциям из educational-tasks.md
-```
-
-Удачи в изучении Apache Airflow!
-
----
-
-**Примечание**: Этот стенд предназначен исключительно для учебных целей. Для production использования требуется дополнительная настройка безопасности.
+Повторный запуск - `docker compose up -d`. Для полного сброса обеих БД используйте `docker compose down -v`: эта команда удалит учебные таблицы, историю запусков и настройки Airflow в томах. Файлы в `data/` останутся.
## Для авторов курса
-[План обновления практики](../docs/specs/2026-09-28-practice-improvement-plan.md) — согласованный подход, найденные несогласованности, порядок доработки и проверки заданий.
+[План обновления практики](../docs/specs/2026-09-28-practice-improvement-plan.md) сохраняет согласованные решения, разбор исходных заданий и результаты проверок.
diff --git a/airflow-docker/dag-specifications.md b/airflow-docker/dag-specifications.md
index ab674da..e6f917c 100644
--- a/airflow-docker/dag-specifications.md
+++ b/airflow-docker/dag-specifications.md
@@ -1,327 +1,29 @@
-# Educational DAG Specifications for Airflow Learning
+# Что делают исходные DAG
-## Learning Progression Structure
+Это справочник примеров для Airflow 2.9.2. Порядок прохождения, изменения кода и условия завершения находятся в [практических заданиях](educational-tasks.md).
-### Level 1: Basic Concepts (Week 1)
+| DAG и файл в `dags/` | Подготовка | Поведение и результат |
+|---|---|---|
+| `hello_world_dag` (`hello_world_dag.py`) | Не требуется | Три Python-задачи печатают приветствие, дату и завершение. Расписание ежедневное, `catchup=False`. |
+| `sql_basic_dag` (`sql_basic_dag.py`) | Connection `postgres_training`, создаваемый при запуске стенда | Создает `public.students_sample`, очищает перед вставкой, добавляет трех студентов и выполняет SELECT. После каждого полного запуска в таблице три строки. |
+| `file_operations_dag` (`file_operations_dag.py`) | Каталоги `data/input`, `data/output` из репозитория | Генерирует 100 строк, проверяет данные, сохраняет `output/processed_data.csv` и `output/summary.txt`. Категория High начинается с 70000. |
+| `csv_to_postgres` (`csv_to_postgres.py`) | Connection `postgres_training` | Создает `public.orders`, генерирует CSV на 1000 строк, передает путь через XCom и загружает файл. Уже существующие `order_id` пропускаются; новый полный запуск генерирует новые заказы. |
+| `csv_to_postgres_dq` (`csv_to_postgres_dq.py`) | Успешная загрузка `csv_to_postgres`; запускать вручную после нее | Проверяет таблицу, точную схему из четырех колонок, наличие строк и отсутствие дубликатов. Сводка перечисляет состояния проверок. По умолчанию после ошибки последующие задачи не выполняются (`all_success`). |
+| `data_processing_dag` (`data_processing_dag.py`) | Не требуется, свои входные CSV создаются заново | Извлекает 5 клиентов и 6 заказов в `output/extracted_*.csv`, читает их при преобразовании. `simulate_load` печатает число строк и DDL без записи в БД. Отчет: 6 заказов, выручка 2105. |
+| `branching_dag` (`branching_dag.py`) | Variable `branch_format`: `csv` или `json`; по умолчанию `csv` | Выбранная ветка имитирует обработку, другая получает `skipped`. Слияние читает XCom текущего запуска с `none_failed_min_one_success`. Неизвестный формат приводит к ошибке. |
+| `error_handling_dag` (`error_handling_dag.py`) | Variable `error_mode`, по умолчанию `retry_once` | Две рабочие задачи, по одному повтору с задержкой 10 секунд. Оба обработчика зависят от обеих задач. Окончательный отказ сохраняет `failed` у DAG Run, даже если обработчик ошибки успешен. |
+| `resource_management_dag` (`resource_management_dag.py`) | Создать `training_pool` с 2 слотами | Две задачи по 20 секунд имитируют чтение и занимают 2 и 1 слот, поэтому работают последовательно. Через XCom передают 20 и 30; следующая задача возвращает словарь, последняя печатает его. |
-#### 1.1 hello_world_dag.py
-**Learning Objectives:**
-- Understand basic DAG structure
-- Learn about PythonOperator
-- Understand task dependencies
+Все DAG, кроме `hello_world_dag`, запускаются вручную. Связи между загрузчиком CSV и DQ-DAG нет. Файлы из `input/orders.csv` в ETL-примере и таблица `public.orders` в загрузчике относятся к разным блокам.
-**DAG Structure:**
-```python
-from airflow import DAG
-from airflow.operators.python import PythonOperator
-from datetime import datetime
+## Режимы примера с ошибками
-def print_hello():
- print("Hello World from Airflow!")
+| `error_mode` | Рабочие задачи | Обработчики | DAG Run |
+|---|---|---|---|
+| `success` | Обе успешны с первой попытки | Успех: `success`, ошибка: `skipped` | `success` |
+| `retry_once` | `retry_task` падает один раз, затем успешна | Успех: `success`, ошибка: `skipped` | `success` |
+| `fail_first` | `unreliable_task` окончательно падает | Успех: `upstream_failed`, ошибка: `success` | `failed` |
+| `fail_second` | `retry_task` окончательно падает | Успех: `upstream_failed`, ошибка: `success` | `failed` |
+| `fail_both` | Обе окончательно падают | Успех: `upstream_failed`, ошибка: `success` | `failed` |
-def print_date():
- print(f"Current date: {datetime.now()}")
-
-def print_goodbye():
- print("Goodbye from Airflow!")
-
-# DAG definition with simple tasks
-```
-
-**Tasks:**
-- `start_task`: Print welcome message
-- `date_task`: Print current date/time
-- `end_task`: Print goodbye message
-
-**Dependencies:**
-start_task → date_task → end_task
-
-#### 1.2 sql_basic_dag.py
-**Learning Objectives:**
-- Connect to PostgreSQL database
-- Execute SQL queries
-- Use PostgresOperator
-
-**Connection Hint:** `docker-compose run --rm airflow-init` automatically provisions the `postgres_training` connection via `airflow connections add`, so no manual setup is required. You can verify it with `docker-compose exec airflow-webserver airflow connections get postgres_training`.
-
-**Tasks:**
-- `create_table`: Create simple table (users, products)
-- `insert_data`: Insert sample records
-- `query_data`: Select and display data
-- `drop_table`: Clean up (optional)
-
-**SQL Operations:**
-```sql
--- Create table
-CREATE TABLE IF NOT EXISTS students (
- id SERIAL PRIMARY KEY,
- name VARCHAR(100),
- age INTEGER,
- created_at TIMESTAMP DEFAULT NOW()
-);
-
--- Insert data
-INSERT INTO students (name, age) VALUES
-('Alice', 22),
-('Bob', 24),
-('Charlie', 21);
-```
-
-### Level 2: Intermediate Concepts (Week 2)
-
-#### 2.1 file_operations_dag.py
-**Learning Objectives:**
-- File system operations
-- CSV data processing
-- Data transformation
-
-**Tasks:**
-- `generate_sample_data`: Create CSV file with random data
-- `read_csv_file`: Read and validate data
-- `transform_data`: Simple data transformations
-- `write_output`: Save processed data
-
-**Sample Data Structure:**
-```csv
-id,name,department,salary
-1,Alice,Engineering,50000
-2,Bob,Marketing,45000
-3,Charlie,Sales,48000
-```
-
-#### 2.2 csv_to_postgres.py
-**Learning Objectives:**
-- Load CSV data into PostgreSQL database
-- Implement idempotent loading via temporary table
-- Use XCom for passing file paths between tasks
-- Work with PostgreSQL connections in Airflow
-
-**Scenario:**
-Generate sample orders data as CSV, preview it, and load it into PostgreSQL.
-
-**Tasks:**
-- `create_orders_table`: Create public.orders table in PostgreSQL
-- `generate_csv`: Generate sample orders CSV file
-- `preview_csv`: Display first few rows of CSV
-- `load_csv_to_postgres`: Load CSV data into PostgreSQL using temporary table
-
-**Database Connection:** Uses `postgres_training` connection (auto-provisioned by init script).
-**Generated Files:** CSV files are written to `/opt/airflow/data/output/`.
-
-**Sample Data Structure:**
-```csv
-order_id,order_ts,customer_id,amount
-1,2023-10-01 10:30:00,101,1250.50
-2,2023-10-01 11:45:00,102,890.00
-3,2023-10-02 09:15:00,103,2100.75
-```
-
-**Data Quality Checks:** Run `csv_to_postgres_dq.py` after loading to validate the target table.
-
-#### 2.3 csv_to_postgres_dq.py
-**Learning Objectives:**
-- Implement data quality validation in Airflow
-- Use Python functions for data checks
-- Handle data quality failures
-- Separate validation from main ETL pipeline
-
-**Scenario:**
-Run automated data quality checks on the public.orders table after CSV loading.
-
-**Tasks:**
-- `check_table_exists`: Verify public.orders table exists
-- `check_schema`: Validate table schema matches expected structure
-- `check_row_count`: Ensure table has data
-- `check_duplicates`: Verify no duplicate order_id values
-
-**Quality Checks:**
-- Table existence in public schema
-- Column names and data types (order_id, order_ts, customer_id, amount)
-- Minimum row count (> 0)
-- Unique order_id values (no duplicates)
-
-**Helper Functions:** All DQ check functions are defined directly in `csv_to_postgres_dq.py`.
-
-#### 2.4 data_processing_dag.py
-**Learning Objectives:**
-- ETL pipeline concepts
-- Multiple data sources
-- Error handling basics
-
-**Tasks:**
-- `extract_customers`: Read customer data
-- `extract_orders`: Read order data
-- `transform_data`: Join and process data
-- `load_to_database`: Save results
-- `generate_report`: Create summary
-
-### Level 3: Advanced Concepts (Week 3)
-
-#### 3.1 branching_dag.py
-**Learning Objectives:**
-- Conditional task execution
-- BranchPythonOperator
-- Decision making in workflows
-
-**Scenario:**
-Process data based on file type or data quality
-
-**Tasks:**
-- `check_file_type`: Determine processing path
-- `process_csv_branch`: For CSV files
-- `process_json_branch`: For JSON files
-- `merge_results`: Combine outputs
-
-#### 3.2 error_handling_dag.py
-**Learning Objectives:**
-- Task retries
-- Error notifications
-- Failure handling
-
-**Tasks:**
-- `unreliable_task`: Simulate failures
-- `retry_task`: Demonstrate retry mechanism
-- `success_handler`: On success callback
-- `failure_handler`: On failure callback
-
-### Level 4: Orchestration & Collaboration (Week 4)
-
-#### 4.1 advanced_features_dag.py
-**Learning Objectives:**
-- Control concurrency with pools and `pool_slots`
-- Exchange data between tasks using XCom
-- Group related tasks using `TaskGroup`
-- Configure email-based alerting on failures
-
-**Scenario:**
-Enhanced daily analytics pipeline that reads data from the training database, performs transformations, and writes summaries, while limiting heavy backup tasks via a dedicated pool and sending notifications about pipeline status.
-
-**Tasks:**
-- `extract_group`: Use `TaskGroup` to wrap extract tasks (e.g., customers and orders)
-- `transform_group`: Aggregate metrics and prepare summary tables
-- `load_group`: Simulate loading results back into the training database or files
-- `backup_task`: Heavy backup task running in a dedicated pool (e.g., `backup_pool`) with custom `pool_slots`
-- `calculate_metrics`: Python task that returns aggregated metrics (pushed to XCom)
-- `log_metrics`: Task that reads metrics via `xcom_pull` and logs them or uses them in a template
-- `send_notification`: Final notification task (email or log) triggered with `ALL_DONE` semantics
-
-## Sample Data Files
-
-### customers.csv
-```csv
-customer_id,name,email,join_date
-1,Alice Johnson,alice@example.com,2023-01-15
-2,Bob Smith,bob@example.com,2023-02-20
-3,Charlie Brown,charlie@example.com,2023-03-10
-```
-
-### orders.csv
-```csv
-order_id,customer_id,product,amount,order_date
-101,1,Laptop,1200,2023-10-01
-102,2,Monitor,300,2023-10-02
-103,1,Keyboard,80,2023-10-03
-104,3,Mouse,25,2023-10-04
-```
-
-### products.csv
-```csv
-product_id,name,category,price
-1,Laptop,Electronics,1200
-2,Monitor,Electronics,300
-3,Keyboard,Electronics,80
-4,Mouse,Electronics,25
-```
-
-## Database Schema for Training
-
-### Students Table
-```sql
-CREATE TABLE students (
- student_id SERIAL PRIMARY KEY,
- first_name VARCHAR(50),
- last_name VARCHAR(50),
- email VARCHAR(100),
- enrollment_date DATE,
- grade INTEGER
-);
-```
-
-### Courses Table
-```sql
-CREATE TABLE courses (
- course_id SERIAL PRIMARY KEY,
- course_name VARCHAR(100),
- instructor VARCHAR(100),
- credits INTEGER
-);
-```
-
-### Enrollments Table
-```sql
-CREATE TABLE enrollments (
- enrollment_id SERIAL PRIMARY KEY,
- student_id INTEGER REFERENCES students(student_id),
- course_id INTEGER REFERENCES courses(course_id),
- enrollment_date DATE,
- grade CHAR(1)
-);
-```
-
-## Learning Outcomes by Week
-
-### Week 1: Foundation
-- ✅ Understand DAG structure and components
-- ✅ Create basic Python tasks
-- ✅ Set up task dependencies
-- ✅ Run first successful workflow
-
-### Week 2: Integration
-- ✅ Connect to databases
-- ✅ Execute SQL operations
-- ✅ Process file data
-- ✅ Build simple ETL pipelines
-
-### Week 3: Advanced Features
-- ✅ Implement conditional logic
-- ✅ Handle errors and retries
-- ✅ Use parameters and templates
-- ✅ Monitor and debug workflows
-
-### Week 4: Orchestration & Operations
-- ✅ Use pools to control resource usage
-- ✅ Share data between tasks via XCom
-- ✅ Group tasks using `TaskGroup`
-- ✅ Configure alerting and notifications for failures
-
-## Common Pitfalls and Solutions
-
-### Problem: DAG not appearing in UI
-**Solution:** Check DAG file location and syntax
-
-### Problem: Database connection errors
-**Solution:** Verify connection strings and database availability
-
-### Problem: Task failures
-**Solution:** Check logs, implement proper error handling
-
-### Problem: Scheduling issues
-**Solution:** Understand cron expressions and execution dates
-
-## Assessment Criteria
-
-### Basic Competency
-- Can create simple DAG with 3+ tasks
-- Understands task dependencies
-- Can run and monitor workflows
-
-### Intermediate Competency
-- Can integrate with databases
-- Can process file data
-- Implements basic error handling
-
-### Advanced Competency
-- Uses conditional branching
-- Implements proper error handling
-- Creates reusable components
-- Optimizes workflow performance
+Меняйте Variable после завершения предыдущего запуска. `retries=1` означает две попытки всего. `failure_handler` использует `one_failed`: достаточно одного окончательного отказа, ждать завершения другой рабочей задачи ему не обязательно.
diff --git a/airflow-docker/educational-tasks.md b/airflow-docker/educational-tasks.md
index a40dbad..8599f62 100644
--- a/airflow-docker/educational-tasks.md
+++ b/airflow-docker/educational-tasks.md
@@ -1,478 +1,435 @@
-# Учебные задания по Apache Airflow
+# Практика по Apache Airflow
-Этот документ содержит практические задания для закрепления знаний по Apache Airflow. Каждое задание соответствует одному из учебных DAG'ов и направлено на лучшее понимание конкретных концепций.
+Здесь мы знакомимся с Airflow на готовых примерах: запускаем DAG, меняем задачи и смотрим, что произошло. Свой цельный процесс обработки данных вы соберете в курсовых работах дальше по программе.
-## Общие инструкции
+## Как проходить практику
-Перед выполнением заданий:
-1. Убедитесь, что стенд Airflow запущен: `docker-compose up -d`
-2. Проверьте доступность интерфейса: http://localhost:8080
-3. Ознакомьтесь с соответствующим DAG'ом в интерфейсе Airflow
-4. Откройте исходный код DAG'а и прочитайте описание в начале файла
+В каждом блоке сначала выполните основную практику. Задания "По желанию" находятся в конце блока: их можно пропустить и перейти дальше. Сложность указана отдельно. Если дополнительному заданию нужны результаты другого упражнения, это написано в его начале.
----
+Основные задания одного блока выполняйте последовательно, сохраняя изменения. Перед новым блоком прочитайте условия начала. Загрузка CSV и проверки DQ используют одну таблицу, поэтому идут вместе.
-## Задания для hello_world_dag.py
+| Блок | Примеры | Что наблюдаем | Перед практикой |
+|---|---|---|---|
+| [1. Первый DAG](#first-dag) | `hello_world_dag.py` | Задачи, зависимости, операторы, расписание | Главы 4, 5 и 8 |
+| [2. SQL и подключения](#sql) | `sql_basic_dag.py` | Connection, SQL, ошибка подключения | Глава 7 |
+| [3. Работа с файлами](#files) | `file_operations_dag.py` | Результаты отдельных задач | Главы 5 и 6 |
+| [4. Загрузка и DQ](#load-dq) | `csv_to_postgres.py`, `csv_to_postgres_dq.py` | XCom, данные в БД, ошибка проверки | Главы 6, 7 и 9 |
+| [5. Разбор ETL](#etl) | `data_processing_dag.py` | Параллельные шаги и путь данных | Глава 5 |
+| [6. Ветвление](#branching) | `branching_dag.py` | Variables, выбранная ветка, `skipped` | Главы 6 и 7 |
+| [7. Ошибки и повторы](#errors) | `error_handling_dag.py` | Попытки, обработчики, итог запуска | Главы 6 и 9 |
+| [8. Пулы и группировка](#resources) | `resource_management_dag.py` | Слоты, XCom, TaskGroup | Глава 9 |
-**Цель:** Освоить базовые концепции Airflow — создание задач, зависимости, операторы.
+Главы доступны в [оглавлении учебника](../README.md). Инструкция запуска и параметры учебной БД находятся в [README стенда](README.md).
-### Задание 1: Модификация существующих задач
-**Сложность:** Начальная
+Откройте http://localhost:8080, войдите как `admin` / `admin`. Для выбранного DAG включите переключатель и нажмите Trigger DAG. Дождитесь завершения запуска, затем откройте его граф и логи задач. После правки Python-файла дождитесь обновления вкладки Code; новый запуск должен использовать новый код. Если DAG исчез или перестал обновляться, проверьте Import Errors и логи scheduler.
-**Задача:**
-- Откройте файл `hello_world_dag.py`
-- Измените текст в функциях `print_hello()`, `print_date()`, `print_goodbye()` на русский язык
-- Добавьте новую задачу, которая выводит текущую погоду (можно использовать фиктивные данные)
+Каталог `/opt/airflow/data/` в контейнере соответствует `airflow-docker/data/` на компьютере. Команды ниже выполняются из `airflow-docker`. Подойдет терминал Linux или PowerShell в Windows.
-**Подсказка:**
-- Новая задача создаётся по аналогии с `date_task` — нужна Python-функция и `PythonOperator` для неё
-- Не забудьте встроить задачу в цепочку зависимостей (`>>`)
-- После сохранения файла DAG подхватится автоматически — запустите его вручную через UI и проверьте логи
+Для SQL можно открыть консоль учебной БД:
-**Цель задания:** Понять структуру DAG, научиться добавлять и модифицировать задачи.
+```bash
+docker compose exec postgres-training psql -U student -d training
+```
-### Задание 2: Изменение расписания
-**Сложность:** Начальная
+В ней выполняются SQL-запросы из заданий; выход - `\q`. Сохраняйте исходные DAG перед опытами вне каталога `dags`: две Python-копии с одинаковым `dag_id` мешают Airflow выбрать нужную.
-**Задача:**
-- Измените `schedule_interval` с ежедневного на еженедельное выполнение
-- Добавьте параметр `max_active_runs` для ограничения одновременных запусков
-- Проверьте изменения в интерфейсе Airflow
+
-**Подсказка:**
-- Расписание задаётся в параметрах `DAG(...)` — ищите `schedule_interval`
-- Можно использовать `timedelta(weeks=1)` или cron-выражение (например, `'0 0 * * 1'`)
-- `max_active_runs` — тоже параметр `DAG()`
+## 1. Первый DAG
-**Цель задания:** Освоить настройку расписания и параметров выполнения DAG.
+Начните с исходного `hello_world_dag.py`. В нем три задачи: `start_task`, `date_task`, `end_task`. Включение DAG с расписанием может создать автоматический запуск; для опытов выбирайте свой ручной запуск.
-### Задание 3: Добавление BashOperator
-**Сложность:** Средняя
+### 1.1. Добавить задачу
-**Задача:**
-- Добавьте задачу с `BashOperator`, которая создает текстовый файл в папке `/opt/airflow/data/output/`
-- Настройте зависимости так, чтобы эта задача выполнялась между `date_task` и `end_task`
-- Убедитесь, что файл создается при каждом запуске DAG
+Основная практика. Сложность: начальная.
-**Подсказка:**
-- Импорт `BashOperator` уже есть в файле — посмотрите шапку
-- `BashOperator` принимает параметр `bash_command` со строкой shell-команды
-- Для записи в файл подойдёт обычный `echo "текст" > /path/to/file.txt`
-- Проверить результат можно: `docker-compose exec airflow-webserver cat /opt/airflow/data/output/<ваш_файл>`
+Запустите исходный DAG и найдите сообщения всех трех задач. Переведите приветствие на русский, затем добавьте функцию с фиктивной погодой и `PythonOperator` с `task_id="weather_task"`. Включите ее между `start_task` и `date_task`, изменив существующую цепочку зависимостей.
-**Цель задания:** Научиться работать с разными типами операторов.
+**Что должно получиться:** в новом запуске четыре задачи со статусом `success`, а в логе `weather_task` есть выбранный вами текст. В графе видно, какая задача должна закончиться перед ней.
----
+### 1.2. Создать файл через BashOperator
-## Задания для sql_basic_dag.py
+Основная практика. Сложность: начальная.
-**Цель:** Освоить работу с базами данных через Airflow.
+Добавьте между `date_task` и `end_task` задачу `write_note`. Импорт `BashOperator` уже есть. В `bash_command` запишите команду `echo 'Airflow works' > /opt/airflow/data/output/note.txt`.
-### Задание 1: Расширение структуры таблицы
-**Сложность:** Средняя
+После запуска проверьте файл:
-**Задача:**
-- Добавьте в таблицу `students_sample` новые поля: `email`, `phone`, `registration_date`
-- Модифицируйте SQL-запросы для работы с новой структурой
-- Добавьте задачу для обновления существующих записей
+```bash
+docker compose exec airflow-webserver cat /opt/airflow/data/output/note.txt
+```
-**Подсказка:**
-- Новые колонки добавляются в переменную `create_table_sql` — подберите подходящие типы (VARCHAR, DATE и т.д.)
-- Не забудьте обновить `insert_data_sql` — количество значений в INSERT должно совпадать с количеством колонок
-- Для задачи UPDATE создайте новую SQL-переменную и новый `PostgresOperator` по аналогии с существующими
-- **Подводный камень:** `TRUNCATE TABLE` не изменяет структуру. Чтобы новые колонки появились, замените его на `DROP TABLE IF EXISTS`
+**Что должно получиться:** пять успешных задач и файл с текстом `Airflow works`. При повторном запуске этот файл перезаписывается.
-**Цель задания:** Научиться работать с миграциями схемы базы данных.
+### 1.3. Посмотреть расписание
-### Задание 2: Создание отчетов
-**Сложность:** Средняя
+Основная практика. Сложность: начальная.
-**Задача:**
-- Добавьте задачу, которая создает сводный отчет по данным студентов
-- Отчет должен содержать: количество студентов, средний возраст, распределение по возрасту
-- Сохраните отчет в файл в папке `/opt/airflow/data/output/`
+Замените `schedule_interval` на `'0 0 * * 1'`. Найдите новое расписание и следующий автоматический запуск в UI. Сопоставьте дату запуска с часовым поясом DAG. `catchup=False` оставьте как есть.
-**Подсказка:**
-- Здесь понадобится `PythonOperator`, а не `PostgresOperator` — потому что нужно получить результат SQL и записать в файл
-- Для подключения к БД из Python-функции используйте `PostgresHook` (импорт: `from airflow.providers.postgres.hooks.postgres import PostgresHook`)
-- У хука есть метод `get_conn()`, который возвращает обычное psycopg2-соединение — дальше работайте через `cursor.execute()` и `fetchone()`
-- Задачу поставьте перед `drop_table` в цепочке зависимостей
+**Что должно получиться:** UI показывает запуск по понедельникам в 00:00 в часовом поясе DAG. Ручной запуск по кнопке по-прежнему выполняется сразу, когда доступны ресурсы. Для проверки расписания ждать понедельника не нужно. После опыта выключите переключатель DAG.
-**Цель задания:** Освоить создание аналитических отчетов в DAG'ах.
+### 1.A. Ограничить одновременные запуски
-### Задание 3: Работа с соединениями
-**Сложность:** Средняя
+По желанию. Сложность: средняя.
-**Задача:**
-- Создайте новое соединение к базе данных через Airflow UI
-- Модифицируйте DAG для использования нового соединения
-- Добавьте обработку ошибок подключения к БД
+В `DAG(...)` задайте `max_active_runs=1`. В одну Python-функцию временно добавьте `time.sleep(30)` с импортом `time`. Включите DAG и запустите его вручную дважды с разными Run Id.
-**Подсказка:**
-- Соединения создаются в UI: Admin → Connections → "+"
-- Параметры для нашей учебной БД: Host = `postgres-training`, Schema = `training`, Login = `student`, Password = `student`, Port = `5432`
-- В DAG-файле соединение указывается через `postgres_conn_id` — замените значение на ваш новый Conn Id
-- Для проверки ошибок попробуйте указать несуществующее соединение и посмотрите, что покажут логи
+**Что должно получиться:** пока первый запуск выполняет задержку, второй не начинает свои задачи. После первого он продолжает работу. Удалите задержку и выключите DAG; ограничение можно оставить.
-**Цель задания:** Научиться управлять соединениями с внешними системами.
+
----
+## 2. SQL и подключения
-## Задания для file_operations_dag.py
+Используйте исходный `sql_basic_dag.py`. Он создает `public.students_sample`, очищает ее перед вставкой и добавляет трех студентов. После выполнения данные остаются для проверки. Повторный полный запуск снова оставит три строки. Остальные таблицы учебной БД этот DAG не меняет.
-**Цель:** Освоить работу с файлами и данными в Airflow.
+### 2.1. Подключиться к учебной БД
-### Задание 1: Модификация генерации данных
-**Сложность:** Средняя
+Основная практика. Сложность: начальная.
-**Задача:**
-- Добавьте новые поля в генерируемые данные: `department`, `experience_years`, `education_level`
-- Модифицируйте валидацию для проверки новых полей
-- Добавьте фильтрацию данных по определенным критериям (например, опыт > 3 года)
+Запустите 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.
-**Подсказка:**
-- Генерация данных происходит в функции `generate_sample_data()` — новые поля добавляются в словарь `data`
-- Для `department` подойдёт `random.choice()` со списком отделов, для `experience_years` — привяжите к возрасту (не может быть больше `age - 18`)
-- Валидация — в функции `read_and_validate_data()`: добавьте `assert` по аналогии с существующими
-- Фильтрацию (pandas `df[df['experience_years'] > 3]`) можно добавить в `transform_data()` и сохранить отдельным файлом
+Проверьте таблицу:
-**Цель задания:** Научиться работать с различными типами данных и валидацией.
+```sql
+SELECT name, age FROM public.students_sample ORDER BY name;
+```
-### Задание 2: Создание дополнительных отчетов
-**Сложность:** Средняя
+**Что должно получиться:** Alice - 22, Bob - 24, Charlie - 21. Новый запуск с вашим Connection также успешен.
-**Задача:**
-- Создайте отчет в формате JSON с детальной статистикой по отделам
-- Добавьте визуализацию данных с помощью библиотеки matplotlib (сохранение графика в файл)
-- Создайте сводку по зарплатам в разных возрастных категориях
+Чтобы увидеть ошибку подключения, временно укажите несуществующий Conn Id и поставьте `retries=0` в `default_args`. Запустите DAG и найдите причину в логе первой задачи. Исправьте Conn Id, верните исходное число повторов и создайте новый запуск. Ручная отметка задачи как успешной не проверяет подключение.
-**Подсказка:**
-- Создайте новую функцию и новый `PythonOperator`, добавьте в цепочку после `summary_task`
-- Для JSON-отчёта: загрузите processed_data.csv через pandas, сгруппируйте (`groupby`) и сохраните через `json.dump()`
-- matplotlib может быть не установлен в контейнере — это нормально, сделайте эту часть необязательной (обработайте `ImportError`)
-- Не забудьте `import pandas as pd` внутри функции (не в шапке файла!)
+### 2.2. Добавить SQL-задачу
-**Цель задания:** Освоить создание комплексных отчетов и визуализацию.
+Основная практика. Сложность: средняя.
-### Задание 3: Оптимизация обработки
-**Сложность:** Продвинутая
+Добавьте `PostgresOperator` с `task_id="update_student"` между `insert_data` и `query_data`. Его SQL должен установить Alice возраст 25. Используйте то же соединение, что и остальные задачи.
-**Задача:**
-- Разделите обработку данных на параллельные задачи для разных отделов
-- Добавьте контроль качества данных (проверка на дубликаты, аномалии)
-- Реализуйте механизм повторной обработки при ошибках
+**Что должно получиться:** после запуска у Alice возраст 25, остальные строки не изменились. Повторный полный запуск дает те же три строки: сначала DAG очищает таблицу, потом вставляет данные и выполняет UPDATE.
-**Подсказка:**
-- Параллельные задачи задаются списком: `read_task >> [task_a, task_b, task_c] >> summary_task`
-- Каждая задача фильтрует DataFrame по своему отделу и сохраняет результат в отдельный файл
-- Проверку дубликатов можно сделать через `df['id'].is_unique`
-- Параметры `retries` и `retry_delay` можно задать как у отдельных задач, так и в `default_args`
+### 2.A. Добавить колонку
-**Цель задания:** Научиться оптимизировать и делать обработку данных отказоустойчивой.
+По желанию. Сложность: средняя. Начните после 2.2.
----
+Добавьте `email VARCHAR(100)` в `create_table_sql` и заполните его в `insert_data_sql`. Для уже созданной таблицы сначала выполните:
-## Задания для csv_to_postgres.py
+```sql
+ALTER TABLE public.students_sample ADD COLUMN email VARCHAR(100);
+```
-**Цель:** Освоить паттерны загрузки данных (ETL) из файлов в базу данных и применение XCom.
+Команда нужна один раз. `CREATE TABLE IF NOT EXISTS` не меняет существующую схему, а `TRUNCATE` очищает только строки.
-### Задание 1: Расширение структуры данных
-**Сложность:** Начальная
+**Что должно получиться:** после запуска в таблице три строки с непустым `email`. Задание 2.B можно выполнять и без этой колонки. Удаление таблицы или всей базы для опыта не требуется.
-**Задача:**
-- Добавьте в генератор CSV новую колонку `status` (например, со случайными значениями 'NEW', 'PROCESSING', 'COMPLETED').
-- Обновите функцию `_create_table`, чтобы учесть новую колонку.
-- Запустите DAG и проверьте, что данные успешно загрузились с новой колонкой.
-- Убедитесь, что сгенерированный файл появился в каталоге `/opt/airflow/data/output/`.
+### 2.B. Сохранить отчет
-**Подсказка:**
-- Новая колонка добавляется в трёх местах: генерация (`_generate_csv`), DDL (`_create_table`), загрузка (`_load_csv` — список колонок в COPY)
-- В `_generate_csv()` для случайных значений подойдёт `random.choices(['NEW', 'PROCESSING', 'COMPLETED'], k=rows)`
-- **Подводный камень:** если таблица уже существует со старой схемой, новая колонка не появится. Удалите таблицу вручную или добавьте `DROP TABLE` перед `CREATE TABLE`
+По желанию. Сложность: средняя. Требуется 2.2; задание 2.A можно пропустить.
-**Цель задания:** Понять процесс изменения схемы данных на всех этапах пайплайна.
+После `query_data` добавьте Python-задачу. Через `PostgresHook` получите количество студентов, средний возраст и количество студентов каждого возраста. Запишите результат в `/opt/airflow/data/output/students_report.txt`. У хука есть методы `get_first()` и `get_records()`; они позволяют обойтись без ручной работы с курсором.
-### Задание 2: Использование PostgresOperator
-**Сложность:** Средняя
+**Что должно получиться:** в файле 3 студента, средний возраст около 23.33 и по одному студенту возрастов 21, 24 и 25. Эти числа должны совпасть с SQL по `students_sample`.
-**Задача:**
-- Перепишите задачу `create_orders_table`. Сейчас она использует `PythonOperator` и `PostgresHook` внутри Python-функции.
-- Замените её на использование стандартного `PostgresOperator`, используя соединение `postgres_training`.
-- Убедитесь, что пайплайн продолжает работать корректно.
+
-**Подсказка:**
-- Посмотрите, как устроен `sql_basic_dag.py` — там `PostgresOperator` уже используется, можно взять за образец
-- `PostgresOperator` принимает `postgres_conn_id` и `sql` — SQL можно передать прямо строкой
-- После замены проверьте, не осталось ли ссылок на удалённую функцию. `_get_conn()` всё ещё нужна в `_load_csv()`
+## 3. Работа с файлами
-**Цель задания:** Научиться использовать специализированные операторы для работы с БД вместо кастомного Python-кода.
+Начните с исходного `file_operations_dag.py`. Он сам создает 100 строк в `input/sample_data.csv`; данные каждого полного запуска могут отличаться. Чужие входные файлы для этого блока не нужны.
----
+### 3.1. Изменить преобразование
-## Задания для csv_to_postgres_dq.py
+Основная практика. Сложность: начальная.
-**Цель:** Освоить подходы к обеспечению качества данных (Data Quality) в Airflow.
+Запустите DAG, откройте `data/output/processed_data.csv` и `summary.txt`. Найдите в `transform_data()` правило вычисления `salary_category`. Измените порог категории High с 70000 на 80000, оставив границу Medium равной 50000. Запустите DAG еще раз.
-### Задание 1: Новая проверка качества
-**Сложность:** Средняя
+**Что должно получиться:** в обработанном CSV 100 строк, у каждой строки с зарплатой от 80000 категория High. Сумма количеств High, Medium и Low в отчете равна 100. Сверяйте отчет с CSV того же запуска: случайные данные не позволяют сравнивать только числа из двух отчетов.
-**Задача:**
-- Добавьте новую функцию проверки прямо в `csv_to_postgres_dq.py`, которая будет убеждаться, что все значения в колонке `amount` строго больше нуля.
-- Добавьте вызов этой функции как новую задачу в DAG `csv_to_postgres_dq`.
-- Встройте новую задачу в общую цепочку выполнения (например, перед `data_quality_summary`).
+### 3.A. Добавить отдел и опыт
-**Подсказка:**
-- Возьмите за образец любую из существующих функций проверки (например, `_check_has_rows`) — паттерн одинаковый: подключиться, выполнить SQL, проверить результат, бросить `ValueError` если не ок
-- SQL для проверки: `SELECT COUNT(*) FROM public.orders WHERE amount <= 0`
-- Не забудьте создать `PythonOperator` и добавить задачу в цепочку
+По желанию. Сложность: средняя.
-**Цель задания:** Научиться расширять набор проверок качества данных.
+В `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`.
-### Задание 2: Управление статусом при ошибках (Trigger Rules)
-**Сложность:** Продвинутая
+**Что должно получиться:** основной файл по-прежнему содержит 100 строк, опыт у всех допустимый, а в `experienced.csv` нет опыта 3 года и меньше.
-**Задача:**
-- Смоделируйте ошибку (например, временно измените данные так, чтобы проверки не прошли).
-- По умолчанию, если падает одна проверка, следующие не выполняются (поведение `all_success`).
-- Измените параметры задач так (с помощью `trigger_rule`), чтобы выполнялись *все* проверки, даже если некоторые из них упали.
-- Сделайте так, чтобы задача `data_quality_summary` могла анализировать статусы предыдущих задач и отражать общий итог.
+### 3.B. Отчет по отделам
-**Подсказка:**
-- Для моделирования ошибки: измените `EXPECTED_ORDERS_SCHEMA` так, чтобы схема не совпала. Запустите — увидите, что все задачи после упавшей будут skipped
-- Ключевой параметр — `trigger_rule` у `PythonOperator`. Значение `"all_done"` означает: «запустись в любом случае, когда все upstream завершились (успешно или нет)»
-- Для анализа статусов в `dq_summary`: функция может принимать `**context` и через `context['ti']` получить информацию о статусах предыдущих задач
-- После экспериментов не забудьте вернуть `EXPECTED_ORDERS_SCHEMA` к правильным значениям
+По желанию. Сложность: средняя. Требуется 3.A.
-**Цель задания:** Освоить продвинутую маршрутизацию статусов задач с помощью `trigger_rule`.
+Добавьте задачу после `write_summary`. Она читает полный `processed_data.csv` и пишет `output/departments.json`: для каждого отдела количество сотрудников и среднюю зарплату. Для расчета подойдет pandas `groupby`, для записи - `to_json`.
----
+**Что должно получиться:** сумма численности отделов равна 100, средние зарплаты совпадают с группировкой исходного CSV. График для этого задания не требуется.
-## Задания для data_processing_dag.py
+### 3.C. Разделить обработку по отделам
-**Цель:** Освоить ETL процессы и работу с бизнес-логикой.
+По желанию. Сложность: высокая. Требуется 3.A, отчет 3.B необязателен.
-### Задание 1: Расширение ETL пайплайна
-**Сложность:** Средняя
+После `transform_data` поставьте три параллельные задачи. Каждая читает полный `processed_data.csv`, отбирает свой отдел, проверяет уникальность `id` и сохраняет отдельный файл. После них добавьте сборку этих трех файлов и проверьте количество строк и уникальность `id` уже в общем результате. Не перезаписывайте исходный файл из параллельных задач.
-**Задача:**
-- Добавьте новый источник данных — файл с информацией о продуктах
-- Создайте задачу для объединения данных о заказах с информацией о продуктах
-- Добавьте расчет общей выручки по продуктам
+Для опыта с повтором одна из задач должна намеренно падать только на первой попытке; используйте `ti.try_number` и `retries=1`. Этот прием разобран в блоке 7.
-**Подсказка:**
-- В `create_sample_data()` добавьте генерацию ещё одного CSV (products.csv) с колонками `product`, `category`, `weight_kg` — значения product должны совпадать с теми, что уже есть в orders
-- Создайте функцию `extract_products()` по аналогии с `extract_customers()`
-- Для объединения в `transform_data()` используйте `pd.merge()` по ключу `product`
-- Новую extract-задачу поставьте параллельно с существующими: `create_data_task >> [extract_customers_task, extract_orders_task, extract_products_task]`
+**Что должно получиться:** объединенный результат содержит те же 100 уникальных `id`. В графе три задачи имеют общего предшественника, но не зависят друг от друга. Повтор не добавляет дубликаты: файл отдела перезаписывается.
-**Цель задания:** Научиться работать с множественными источниками данных.
+
-### Задание 2: Расширение текстового отчёта
-**Сложность:** Средняя
+## 4. Загрузка CSV и проверка качества
-**Задача:**
-- Расширьте функцию `generate_report()` так, чтобы отчёт содержал больше полезной статистики
-- Добавьте в отчёт: топ-3 самых дорогих заказа, распределение заказов по месяцам, среднюю сумму заказа по каждому клиенту
-- Сохраните расширенный отчёт в `/opt/airflow/data/output/detailed_report.txt`
+Здесь работают два DAG и одна таблица `public.orders`. Начните с исходных `csv_to_postgres.py` и `csv_to_postgres_dq.py`. SQL-блок использовал другую таблицу. Сначала выполняйте основные задания, затем расширяйте схему по желанию.
-**Подсказка:**
-- Всё делается в существующей функции `generate_report()` — дополните её
-- Полезные методы pandas: `df.nlargest()`, `df.groupby(...).mean()`, `df.groupby(...).sum()`
-- Можно записать результат в тот же файл или создать отдельный
+### 4.1. Проследить загрузку и XCom
-**Цель задания:** Освоить создание аналитических отчётов с помощью pandas.
+Основная практика. Сложность: начальная.
-### Задание 3: Мониторинг качества данных
-**Сложность:** Продвинутая
+Запустите `csv_to_postgres`. В задаче `generate_csv` откройте вкладку XCom с ключом `return_value`: там путь к CSV. Найдите этот файл в `data/output`. Затем выберите `load_csv_to_postgres`, откройте Details и раздел Rendered Templates. В `opKwargs` должен быть тот же путь, подставленный вместо шаблона параметра `csv_path`.
-**Задача:**
-- Добавьте проверки качества данных на каждом этапе ETL
-- Создайте механизм оповещения о проблемах с данными
-- Реализуйте архивирование обработанных данных
+Проверьте результат в БД:
-**Подсказка:**
-- Создайте отдельную функцию валидации (проверка: строки > 0, нет NULL в ключевых полях, суммы положительные) и поставьте её между extract и transform
-- Для оповещения используйте `on_failure_callback` в `default_args` — это функция, которая вызывается при падении любой задачи. Она получает `context` с информацией об ошибке
-- Для архивирования: скопируйте результат в файл с датой в имени (например, `shutil.copy()` + `datetime.now().strftime(...)`)
+```sql
+SELECT COUNT(*) FROM public.orders;
+SELECT order_id, amount FROM public.orders ORDER BY order_id LIMIT 5;
+```
-**Цель задания:** Научиться обеспечивать качество данных в ETL процессах.
+**Что должно получиться:** первый запуск на пустой таблице загружает 1000 строк. В XCom передается путь к файлу, сами строки хранятся в CSV и PostgreSQL. Каждый новый полный запуск генерирует новые заказы; после двух последовательных запусков на изначально пустой таблице будет 2000 строк. Это отличается от повторного выполнения только задачи загрузки с тем же CSV.
----
+### 4.2. Добавить проверку и увидеть ошибку
-## Задания для branching_dag.py
+Основная практика. Сложность: средняя.
-**Цель:** Освоить условную логику и ветвление в Airflow.
+После успешной загрузки отдельно запустите `csv_to_postgres_dq`. Автоматической связи между этими двумя DAG нет.
-### Задание 1: Модификация условий ветвления
-**Сложность:** Средняя
+По образцу `_check_has_rows()` добавьте функцию и Python-задачу `check_positive_amount` перед `data_quality_summary`. Проверьте число строк, где `amount <= 0`, и бросьте `ValueError`, если такие строки есть. Новую задачу включите в существующую цепочку.
-**Задача:**
-- Измените условие в функции `check_data_quality()` на основе реальных критериев (например, размер файла)
-- Добавьте третью ветку обработки для данных "требующих ручной проверки"
-- Настройте разные триггерные правила для слияния веток
+**Что должно получиться на корректных данных:** все задачи и DAG Run успешны, сводка перечисляет результаты проверок.
-**Подсказка:**
-- Сейчас функция выбирает ветку случайно — замените `random.random()` на проверку чего-то реального (например, `os.path.getsize()` для размера файла)
-- Функция `BranchPythonOperator` возвращает `task_id` ветки для выполнения. Для третьей ветки — верните третий `task_id`
-- У `merge_task` проверьте `trigger_rule` — при ветвлении непройденные ветки получают статус `skipped`
+Для отрицательного сценария поменяйте знак суммы одного заказа:
-**Цель задания:** Научиться создавать сложные условия ветвления.
+```sql
+UPDATE public.orders SET amount = -amount
+WHERE order_id = (SELECT MIN(order_id) FROM public.orders);
+```
-### Задание 2: Реализация реального сценария
-**Сложность:** Продвинутая
+Запустите только DQ-DAG. Новая проверка должна получить `failed`, сводка - `upstream_failed`, весь DAG Run - `failed`. В ее логе должно быть понятно, какое условие нарушено. У DQ-задач нет повторов: повторный запуск с теми же неверными данными не исправляет ошибку.
-**Задача:**
-- Создайте реальные задачи обработки для CSV и JSON форматов
-- Добавьте валидацию данных в каждой ветке
-- Реализуйте механизм сравнения результатов из разных веток
+Восстановите сумму тем же UPDATE еще один раз и повторите DQ. Дождитесь успешного результата перед продолжением. Во время опыта не запускайте загрузчик и не удаляйте строки: восстановление относится к тому же минимальному `order_id`.
-**Подсказка:**
-- Вместо `print("Обработка CSV...")` загрузите реальный файл (например, `sample_data.csv`) и посчитайте статистику
-- Результат каждой ветки сохраните в отдельный файл — в `merge_results()` загрузите оба (если существуют) и сравните
-- Учтите, что при ветвлении выполняется только одна ветка — функция слияния должна обрабатывать случай, когда один из файлов отсутствует
+### 4.A. Провести новое поле через загрузку
-**Цель задания:** Применить ветвление в реальном сценарии обработки данных.
+По желанию. Сложность: высокая. Требуется завершенный основной блок 4.
-### Задание 3: Динамическое ветвление
-**Сложность:** Продвинутая
+Добавьте `status` со значениями `NEW`, `PROCESSING`, `COMPLETED`. Поле нужно провести через четыре участка: генератор CSV, DDL целевой таблицы, список колонок COPY и обе части итогового `INSERT ... SELECT` в `_load_csv()`.
-**Задача:**
-- Реализуйте ветвление на основе внешних параметров (например, переданных через Variables)
-- Добавьте обработку случая, когда ни одна ветка не подходит
-- Создайте механизм логирования выбранного пути выполнения
+В DDL используйте `status VARCHAR(20) NOT NULL DEFAULT 'NEW'`. Существующую таблицу подготовьте один раз:
-**Подсказка:**
-- Variables создаются в UI: Admin → Variables. Для чтения в коде: `Variable.get('key', default_var='значение')`
-- Для случая «ни одна ветка не подходит» добавьте `DummyOperator` как ветку-заглушку
-- `BranchPythonOperator` должен всегда возвращать существующий `task_id` — иначе DAG упадёт с ошибкой
+```sql
+ALTER TABLE public.orders
+ADD COLUMN status VARCHAR(20) NOT NULL DEFAULT 'NEW';
+```
-**Цель задания:** Освоить динамическое принятие решений в DAG'ах.
+В `EXPECTED_ORDERS_SCHEMA` DQ-DAG добавьте в конец `("status", "character varying")`. После загрузки проверьте значения `status` в БД и выполните DQ.
----
+**Что должно получиться:** новые заказы содержат статусы из CSV, старые получили `NEW`, проверка схемы проходит. Проверяйте значения, а не только успешность задачи: забытый `status` в INSERT может оставить всем строкам значение по умолчанию.
-## Задания для error_handling_dag.py
+Завершив опыт, верните исходные четыре списка/описания колонок в загрузчике и список схемы в DQ. Затем выполните `ALTER TABLE public.orders DROP COLUMN status;` и запустите DQ. Проверку положительной суммы из 4.2 сохраните.
-**Цель:** Освоить обработку ошибок и создание отказоустойчивых пайплайнов.
+### 4.B. Заменить PythonOperator на PostgresOperator
-### Задание 1: Настройка стратегий повторения
-**Сложность:** Средняя
+По желанию. Сложность: средняя. Используйте базовую схему после 4.2 или после восстановления в 4.A.
-**Задача:**
-- Измените параметры `retries` и `retry_delay` для разных задач
-- Добавьте экспоненциальную задержку между повторными попытками
-- Реализуйте кастомный обработчик ошибок для конкретных исключений
+Замените задачу `create_orders_table` на `PostgresOperator` с тем же `task_id`, SQL создания таблицы и соединением `postgres_training`. Образец оператора есть в SQL-блоке. `_get_conn()` оставьте: загрузчик продолжает использовать эту функцию.
-**Подсказка:**
-- Параметры повторения можно задать на уровне задачи (перекроют `default_args`): `retries`, `retry_delay`, `retry_exponential_backoff`, `max_retry_delay`
-- Для кастомного обработчика: параметр `on_retry_callback` принимает функцию с аргументом `context`, из которого можно получить номер попытки через `context['ti'].try_number`
-- Запустите DAG несколько раз и посмотрите в логах, как меняется задержка между попытками
+**Что должно получиться:** после замены DAG загружает новую порцию, затем DQ проходит. Имя задачи и цепочка зависимостей сохраняются.
-**Цель задания:** Научиться настраивать стратегии обработки ошибок.
+### 4.C. Выполнить все проверки после ошибки
-### Задание 2: Создание комплексной обработки ошибок
-**Сложность:** Продвинутая
+По желанию. Сложность: высокая. Требуется 4.2.
-**Задача:**
-- Добавьте задачи для разных типов ошибок (сетевая ошибка, ошибка данных, системная ошибка)
-- Создайте механизм эскалации ошибок (после N неудачных попыток)
-- Реализуйте отправку уведомлений о критических ошибках
+Временно замените ожидаемый тип `amount` в `EXPECTED_ORDERS_SCHEMA` на `integer`. Сначала запустите исходную цепочку: `check_orders_schema` упадет, последующие задачи получат `upstream_failed`.
-**Подсказка:**
-- Создайте функции по аналогии с `unreliable_task()`, но бросающие разные типы исключений: `ConnectionError`, `ValueError`, `OSError`
-- Для эскалации используйте `on_failure_callback` — внутри проверьте `context['ti'].try_number` и при достижении порога выведите сообщение уровня CRITICAL
-- Пример структуры callback:
- ```python
- def escalation_callback(context):
- if context['ti'].try_number >= 3:
- # сюда — логику эскалации
- ```
+У задач после проверки схемы, включая сводку, задайте `trigger_rule="all_done"`. Повторите запуск. Прочитайте `_log_dq_summary()`: она собирает состояния всех проверок и сама падает, если хотя бы одна неуспешна.
-**Цель задания:** Освоить создание комплексной системы обработки ошибок.
+**Что должно получиться:** проверки строк, дубликатов и положительной суммы выполняются даже после ошибки схемы. Сводка и DAG Run остаются `failed`. Успешная завершающая задача с `all_done` могла бы скрыть отказ внутри DAG; именно поэтому сводка проверяет результат.
-### Задание 3: Добавление логирования через модуль logging
-**Сложность:** Средняя
+Верните тип `numeric` и исходные `trigger_rule`, затем выполните успешный DQ-запуск.
-**Задача:**
-- Замените все `print()` в функциях DAG'а на вызовы стандартного модуля `logging`
-- Добавьте в каждую функцию логирование начала и окончания выполнения
-- Посмотрите, как логи отображаются в интерфейсе Airflow (вкладка Log у каждой задачи)
+
-**Подсказка:**
-- Стандартный паттерн: `import logging` в начале файла, затем `log = logging.getLogger(__name__)` в функции
-- Уровни: `log.info()` для штатных событий, `log.warning()` для предупреждений, `log.error()` для ошибок
-- В UI Airflow откройте выполненную задачу → вкладка **Log** — сообщения `logging` отображаются с метками уровня и временем, в отличие от `print()`
+## 5. Разбор ETL-примера
-**Цель задания:** Научиться использовать стандартное логирование Python в задачах Airflow и читать логи через UI.
+Начните с исходного `data_processing_dag.py`. Он сам создает свои `customers.csv` и `orders.csv`; таблица `public.orders` из блока 4 ему не нужна. Шаг `simulate_load` показывает подготовку к загрузке и печатает пример DDL. Записи в БД в этом DAG нет.
----
+### 5.1. Проследить путь данных
-## Дополнительные задания: пулы, XCom, TaskGroup и алертинг
+Основная практика. Сложность: начальная.
-**Цель:** Освоить продвинутые возможности оркестрации — управление ресурсами, обмен данными между задачами, группировку и оповещения.
+Запустите DAG. Найдите две задачи извлечения и их файлы `output/extracted_customers.csv`, `output/extracted_orders.csv`. Проверьте, что `transform_data()` читает именно их. В графе обе задачи извлечения могут выполняться параллельно, а преобразование ждет обе.
-### Задание 1: Пулы и управление ресурсами
-**Сложность:** Средняя
+В `extract_orders()` временно оставьте только заказы с `amount >= 100` перед сохранением файла. Запустите DAG снова и сравните отчет с результатом до изменения.
-**Задача:**
-- В интерфейсе Airflow создайте пул `backup_pool` с 2 слотами
-- Создайте новый DAG `advanced_features_dag.py` **или** расширьте `data_processing_dag.py` задачами резервного копирования (например, `backup_to_csv`, `backup_to_db`)
-- Назначьте этим задачам параметр `pool="backup_pool"` и настройте `pool_slots` так, чтобы одна из задач занимала 2 слота, а другая — 1
-- Наблюдайте в UI, что одновременно запускается не более 2 задач из этого пула
+**Что должно получиться:** исходный отчет содержит 6 заказов с выручкой 2105; после фильтра - 4 заказа с выручкой 2000. Лог `simulate_load` явно сообщает об имитации. Верните фильтр к исходному состоянию и запустите DAG перед дополнительными заданиями.
-**Подсказка:**
-- Пул создаётся в UI: Admin → Pools → "+"
-- У `PythonOperator` есть параметры `pool` (имя пула) и `pool_slots` (сколько слотов занимает задача)
-- Если задача занимает 2 слота из 2 доступных — вторая задача будет ждать, даже если они не связаны зависимостями
+### 5.A. Добавить справочник продуктов
-**Цель задания:** Научиться управлять параллелизмом задач через пулы и `pool_slots`.
+По желанию. Сложность: средняя.
-### Задание 2: Обмен данными через XCom
-**Сложность:** Средняя
+В `create_sample_data()` добавьте `input/products.csv` с полями `product`, `category`, `weight_kg`. Для каждого из шести продуктов в заказах должна быть одна строка справочника. Добавьте параллельную задачу `extract_products`, сохраняющую `output/extracted_products.csv`, и подключите ее к `transform_data`. Объедините извлеченные заказы со справочником по `product`.
-**Задача:**
-- В том же DAG добавьте задачу `calculate_metrics` (PythonOperator), которая возвращает словарь с агрегированными показателями, например: `{"total_orders": ..., "avg_amount": ...}`
-- Добавьте задачу `log_metrics`, которая с помощью `xcom_pull` читает результат `calculate_metrics` и выводит значения в лог
-- Для одной из задач продемонстрируйте использование XCom в Jinja-шаблоне (например, в `bash_command` или SQL-запросе)
+**Что должно получиться:** в результате 6 заказов, у всех заполнена категория, общая выручка осталась 2105. Если JOIN размножил строки или потерял заказы, проверьте уникальность и полноту справочника.
-**Подсказка:**
-- Любое значение, которое функция возвращает через `return`, автоматически сохраняется в XCom
-- Для чтения XCom в другой задаче: функция принимает `**context`, затем `context['ti'].xcom_pull(task_ids='имя_задачи')`
-- В Jinja-шаблонах (параметры `bash_command`, `sql` и др.) XCom доступен через `{{ ti.xcom_pull(task_ids='...') }}`
-- Посмотреть сохранённые XCom-значения можно в UI: откройте задачу → вкладка **XCom**
+### 5.B. Расширить отчет
-**Цель задания:** Освоить передачу результатов между задачами через XCom и их использование в шаблонах.
+По желанию. Сложность: средняя. Справочник из 5.A не нужен.
-### Задание 3: TaskGroup и алертинг
-**Сложность:** Продвинутая
+В `generate_report()` добавьте топ-3 заказа по сумме, выручку по месяцам и среднюю сумму заказа каждого клиента. Сохраните `output/detailed_report.txt`.
-**Задача:**
-- Объедините логически связанные задачи (например, `extract` / `transform` / `load`) в `TaskGroup`'ы
-- Добавьте завершающую задачу `send_notification` на основе примера из раздела про алертинг (PythonOperator с `send_email_smtp` или другим механизмом уведомлений)
-- Настройте для задачи уведомления `trigger_rule=TriggerRule.ALL_DONE`, чтобы уведомление отправлялось даже при частичных ошибках
-- При желании вынесите функцию отправки письма в отдельный модуль `utils.py` и импортируйте её в DAG
+**Что должно получиться:** топ-3 суммы - 1200, 400 и 300; все заказы относятся к октябрю 2023 года; средняя сумма заказа Alice Johnson равна 640. У исходного набора всего шесть строк, поэтому результат можно сверить вручную.
-**Подсказка:**
-- TaskGroup — контекстный менеджер: `with TaskGroup("имя") as group:` — внутри определяете задачи как обычно
-- Зависимости работают на уровне групп: `group_a >> group_b`
-- `TriggerRule.ALL_DONE` (из `airflow.utils.trigger_rule`) означает: «запустись, когда все upstream завершились — неважно, с успехом или ошибкой»
-- В Graph View группы отображаются как складные блоки — удобно для больших DAG'ов
+### 5.C. Валидация и архив
-**Цель задания:** Научиться группировать задачи с помощью TaskGroup и строить схему оповещений о статусе пайплайна.
+По желанию. Сложность: высокая.
----
+Между преобразованием и `simulate_load` добавьте проверку: результат непустой, `customer_id` и `name` заполнены, суммы положительны. В `create_sample_data()` временно укажите для одного заказа несуществующий `customer_id`. Настройте `on_failure_callback`, который пишет в лог идентификатор задачи и причину ошибки; образец есть в задании 7.B.
-## Рекомендации по выполнению
+После `generate_report` добавьте копирование `transformed_data.csv` в архивный файл с датой и временем в имени. Для этого подойдет `shutil.copy`.
-1. **Начинайте с простых заданий** и постепенно переходите к сложным
-2. **Тестируйте каждое изменение** через интерфейс Airflow
-3. **Изучайте логи выполнения** для понимания поведения задач (вкладка Log у каждой задачи в UI)
-4. **Экспериментируйте** с разными настройками и параметрами
-5. **Не бойтесь ломать** — DAG'и можно откатить через `git checkout`, а таблицы пересоздать
+**Что должно получиться:** после исчерпания повторов проверка падает, последующие шаги получают `upstream_failed`, новый архив не появляется. После возврата клиента к исходному значению весь DAG успешен, а содержимое нового архива совпадает с результатом запуска. Логи callback ищите в логах scheduler, а не только в логе задачи.
-## Оценка прогресса
+
-- **Начальный уровень:** Выполнены задания для hello_world_dag и sql_basic_dag
-- **Средний уровень:** Выполнены задания для file_operations_dag, csv_to_postgres.py, csv_to_postgres_dq.py и data_processing_dag
-- **Продвинутый уровень:** Выполнены все задания, включая branching_dag, error_handling_dag и дополнительные задания по пулам, XCom, TaskGroup и алертингу
+## 6. Ветвление
-Удачи в изучении Apache Airflow!
+Используйте исходный `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 проверяйте в логах scheduler:
+
+```bash
+docker compose logs airflow-scheduler
+```
+
+Если нужно наблюдать каждый повтор, для этого есть отдельный `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. TaskGroup организует граф; ограничения пула продолжают действовать.
+
+### 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.
+
+Задания по желанию расширяют эти опыты. Их выполнение не требуется для перехода к следующему блоку программы.