From 72d5968d01ba174f022f1356db7c9de08f0f71a4 Mon Sep 17 00:00:00 2001
From: Dmitry Dementev <me@dementev.space>
Date: Thu, 26 Mar 2026 22:50:55 +0300
Subject: [PATCH 1/3] =?UTF-8?q?refactor(dags):=20=D0=BF=D0=B5=D1=80=D0=B5?=
 =?UTF-8?q?=D0=BD=D0=B5=D1=81=D1=91=D0=BD=20pandas=20=D0=B2=20=D1=84=D1=83?=
 =?UTF-8?q?=D0=BD=D0=BA=D1=86=D0=B8=D0=B8,=20=D0=B8=D1=81=D0=BF=D1=80?=
 =?UTF-8?q?=D0=B0=D0=B2=D0=BB=D0=B5=D0=BD=20=D0=B1=D0=B0=D0=B3=20occupatio?=
 =?UTF-8?q?ns,=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5=D0=BD=D1=8B=20?=
 =?UTF-8?q?docstring?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Зачем:
  - обратная связь от менти: pandas в шапке DAG замедляет парсинг scheduler,
    отсутствие описаний затрудняет изучение, баг с неиспользуемой переменной.
- Что:
  - import pandas as pd перенесён из шапки внутрь функций (file_operations, csv_to_postgres, data_processing).
  - исправлен баг: переменная occupations в file_operations_dag.py теперь включена в DataFrame.
  - добавлены docstring с описанием шагов DAG для csv_to_postgres.py, csv_to_postgres_dq.py, data_processing_dag.py.
  - расширен docstring file_operations_dag.py.
  - удалён educational-setup-plan.md (технический файл, не для менти).
- Проверка:
  - airflow DagBag: все 8 DAG загружаются без ошибок.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 airflow-docker/dags/csv_to_postgres.py     |  16 ++-
 airflow-docker/dags/csv_to_postgres_dq.py  |  13 ++
 airflow-docker/dags/data_processing_dag.py |  16 ++-
 airflow-docker/dags/file_operations_dag.py |  17 ++-
 airflow-docker/educational-setup-plan.md   | 151 ---------------------
 5 files changed, 54 insertions(+), 159 deletions(-)
 delete mode 100644 airflow-docker/educational-setup-plan.md

diff --git a/airflow-docker/dags/csv_to_postgres.py b/airflow-docker/dags/csv_to_postgres.py
index af7477e..9aa14c2 100644
--- a/airflow-docker/dags/csv_to_postgres.py
+++ b/airflow-docker/dags/csv_to_postgres.py
@@ -1,3 +1,15 @@
+"""
+DAG для загрузки CSV-данных в PostgreSQL
+Уровень: Средний
+
+Этот DAG демонстрирует типичный паттерн загрузки данных из файлов в БД:
+1. Создание целевой таблицы public.orders (если не существует)
+2. Генерация CSV с заказами (order_id, order_ts, customer_id, amount) через pandas
+3. Предпросмотр CSV (head + describe) для контроля
+4. Загрузка в PostgreSQL через временную таблицу и anti-join (защита от дубликатов)
+
+Результат: данные в таблице public.orders учебной БД postgres_training.
+"""
 from __future__ import annotations
 
 import logging
@@ -5,7 +17,7 @@ import os
 import random
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
-import pandas as pd
+
 from airflow.operators.python import PythonOperator
 from airflow.providers.postgres.hooks.postgres import PostgresHook
 
@@ -42,6 +54,7 @@ def _create_table() -> None:
 
 def _generate_csv(rows: int, csv_dir: Path) -> str:
     """Генерирует CSV c заказами с помощью pandas и сохраняет на диск."""
+    import pandas as pd
     csv_dir.mkdir(parents=True, exist_ok=True)
     timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
     csv_path = csv_dir / f"orders_{timestamp}.csv"
@@ -80,6 +93,7 @@ def _generate_csv(rows: int, csv_dir: Path) -> str:
 
 def _preview_csv(csv_path: str, sample_rows: int = 5) -> None:
     """Отображает предпросмотр CSV через pandas (head и describe)."""
+    import pandas as pd
     df = pd.read_csv(csv_path)
     df["order_ts"] = pd.to_datetime(df["order_ts"], errors="coerce")
     logging.info(
diff --git a/airflow-docker/dags/csv_to_postgres_dq.py b/airflow-docker/dags/csv_to_postgres_dq.py
index 38391c5..83822ed 100644
--- a/airflow-docker/dags/csv_to_postgres_dq.py
+++ b/airflow-docker/dags/csv_to_postgres_dq.py
@@ -1,3 +1,16 @@
+"""
+DAG для проверки качества данных (Data Quality) после загрузки в PostgreSQL
+Уровень: Средний
+
+Этот DAG запускается после csv_to_postgres и проверяет таблицу public.orders:
+1. Существование таблицы
+2. Соответствие схемы (колонки и типы данных)
+3. Наличие данных (таблица не пуста)
+4. Отсутствие дубликатов по order_id
+5. Итоговая сводка по результатам проверок
+
+Результат: логи с результатами каждой проверки. При ошибке — ValueError с описанием проблемы.
+"""
 from __future__ import annotations
 
 import logging
diff --git a/airflow-docker/dags/data_processing_dag.py b/airflow-docker/dags/data_processing_dag.py
index da8f57d..4086e84 100644
--- a/airflow-docker/dags/data_processing_dag.py
+++ b/airflow-docker/dags/data_processing_dag.py
@@ -1,12 +1,20 @@
 """
 DAG для демонстрации ETL процессов в Airflow
 Уровень: Средний-Продвинутый
+
+Этот DAG реализует классический ETL-пайплайн:
+1. Создание тестовых данных (клиенты + заказы) в CSV
+2. Параллельное извлечение (Extract) клиентов и заказов
+3. Трансформация (Transform) — объединение таблиц, вычисляемые поля
+4. Загрузка (Load) — подготовка SQL для вставки в БД
+5. Генерация текстового отчёта со статистикой
+
+Результат: файлы в /opt/airflow/data/output/ (extracted_*, transformed_data.csv, report.txt).
 """
 from datetime import datetime, timedelta
 from airflow import DAG
 from airflow.operators.python import PythonOperator
 from airflow.providers.postgres.operators.postgres import PostgresOperator
-import pandas as pd
 
 # Определение DAG
 default_args = {
@@ -30,6 +38,7 @@ dag = DAG(
 
 def create_sample_data():
     """Создание примерных данных для ETL процесса"""
+    import pandas as pd
     # Создаем файлы с данными клиентов и заказов
     customers_data = {
         'customer_id': [1, 2, 3, 4, 5],
@@ -55,6 +64,7 @@ def create_sample_data():
 
 def extract_customers():
     """Извлечение данных клиентов"""
+    import pandas as pd
     df = pd.read_csv('/opt/airflow/data/input/customers.csv')
     print(f"Извлечено {len(df)} записей клиентов")
     
@@ -64,6 +74,7 @@ def extract_customers():
 
 def extract_orders():
     """Извлечение данных заказов"""
+    import pandas as pd
     df = pd.read_csv('/opt/airflow/data/input/orders.csv')
     print(f"Извлечено {len(df)} записей заказов")
     
@@ -73,6 +84,7 @@ def extract_orders():
 
 def transform_data():
     """Преобразование данных - объединение клиентов и заказов"""
+    import pandas as pd
     customers_df = pd.read_csv('/opt/airflow/data/input/customers.csv')
     orders_df = pd.read_csv('/opt/airflow/data/input/orders.csv')
     
@@ -91,6 +103,7 @@ def transform_data():
 
 def load_to_database():
     """Загрузка данных в базу данных (симуляция)"""
+    import pandas as pd
     df = pd.read_csv('/opt/airflow/data/output/transformed_data.csv')
     
     # В реальном сценарии здесь был бы код для загрузки в базу данных
@@ -120,6 +133,7 @@ def load_to_database():
 
 def generate_report():
     """Генерация отчета"""
+    import pandas as pd
     df = pd.read_csv('/opt/airflow/data/output/transformed_data.csv')
     
     # Создаем простой отчет
diff --git a/airflow-docker/dags/file_operations_dag.py b/airflow-docker/dags/file_operations_dag.py
index 9ebc4a4..582989a 100644
--- a/airflow-docker/dags/file_operations_dag.py
+++ b/airflow-docker/dags/file_operations_dag.py
@@ -2,13 +2,14 @@
 DAG для демонстрации работы с файлами в Airflow
 Уровень: Средний
 
-Этот DAG генерирует русскоязычные данные с использованием библиотеки mimesis, включая:
-- ФИО (с учетом пола)
-- Возраст
-- Зарплату
-- Адрес проживания
+Этот DAG показывает полный цикл работы с файлами:
+1. Генерация русскоязычных данных (ФИО, возраст, профессия, зарплата, адрес) с помощью mimesis
+2. Валидация CSV-файла (проверка колонок, диапазонов, пустых значений)
+3. Трансформация (добавление категорий зарплат, возрастных групп, извлечение города)
+4. Формирование текстовой сводки со статистикой
+
+Результат: файлы sample_data.csv, processed_data.csv и summary.txt в /opt/airflow/data/
 """
-import pandas as pd
 import os
 from datetime import datetime, timedelta
 from airflow import DAG
@@ -56,6 +57,7 @@ def generate_sample_data():
         'id': range(1, 101),
         'name': names,
         'age': ages,
+        'occupation': occupations,
         'salary': [max(30000, min(150000, ages[i] * 1200 + (hash(names[i]) % 15000) - 5000)) for i in range(100)],
         'address': [address.address() for _ in range(100)]
     }
@@ -68,6 +70,7 @@ def generate_sample_data():
 
 def read_and_validate_data():
     """Чтение и валидация CSV файла"""
+    import pandas as pd
     df = pd.read_csv('/opt/airflow/data/input/sample_data.csv')
     print(f"Прочитан файл: {len(df)} строк, {len(df.columns)} столбцов")
     print(f"Колонки: {list(df.columns)}")
@@ -87,6 +90,7 @@ def read_and_validate_data():
 
 def transform_data():
     """Преобразование данных"""
+    import pandas as pd
     import os
     df = pd.read_csv('/opt/airflow/data/input/sample_data.csv')
     
@@ -115,6 +119,7 @@ def transform_data():
 
 def write_summary():
     """Создание сводки по обработанным данным"""
+    import pandas as pd
     df = pd.read_csv('/opt/airflow/data/output/processed_data.csv')
     
     summary = {
diff --git a/airflow-docker/educational-setup-plan.md b/airflow-docker/educational-setup-plan.md
deleted file mode 100644
index 40694a3..0000000
--- a/airflow-docker/educational-setup-plan.md
+++ /dev/null
@@ -1,151 +0,0 @@
-# Educational Airflow Setup Plan for Beginners
-
-## Current Issues Identified
-
-### 1. Docker Compose Structure Problems
-- Duplicate `services:` sections in [`docker-compose.yml`](airflow-docker/docker-compose.yml:1,21)
-- Inconsistent container naming
-
-### 2. Missing Directory Structure
-- No `airflow/dags/` directory
-- No `data/` directory for sample files
-- No initialization scripts
-
-## Proposed Simple Educational Setup
-
-### Core Components
-
-#### 1. Fixed Docker Compose Structure
-```yaml
-services:
-  # PostgreSQL for Airflow metadata
-  postgres-metadata:
-    image: postgres:16
-    environment:
-      POSTGRES_USER: airflow
-      POSTGRES_PASSWORD: airflow
-      POSTGRES_DB: airflow
-    ports:
-      - "5434:5432"
-    volumes:
-      - pgmeta:/var/lib/postgresql/data
-
-  # PostgreSQL for training exercises
-  postgres-training:
-    image: postgres:16
-    environment:
-      POSTGRES_USER: student
-      POSTGRES_PASSWORD: student
-      POSTGRES_DB: training
-    ports:
-      - "5432:5432"
-    volumes:
-      - pg_data:/var/lib/postgresql/data
-
-  # Airflow services
-  airflow-webserver:
-    image: apache/airflow:2.9.2
-    environment:
-      AIRFLOW__CORE__LOAD_EXAMPLES: "False"
-      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres-metadata:5432/airflow
-    ports:
-      - "8080:8080"
-    volumes:
-      - ./dags:/opt/airflow/dags
-      - ./data:/opt/airflow/data
-
-  airflow-scheduler:
-    image: apache/airflow:2.9.2
-    environment:
-      AIRFLOW__CORE__LOAD_EXAMPLES: "False"
-      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres-metadata:5432/airflow
-    volumes:
-      - ./dags:/opt/airflow/dags
-      - ./data:/opt/airflow/data
-
-  airflow-init:
-    image: apache/airflow:2.9.2
-    environment:
-      AIRFLOW__CORE__LOAD_EXAMPLES: "False"
-      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres-metadata:5432/airflow
-```
-
-#### 2. Hardcoded Environment Variables
-Create `.env` file with all variables hardcoded:
-- Airflow admin: admin/admin
-- PostgreSQL metadata: airflow/airflow
-- PostgreSQL training: student/student
-
-#### 3. Educational DAG Examples
-
-**Level 1: Basic Concepts**
-- `hello_world_dag.py` - Simple print statements
-- `sql_basic_dag.py` - Basic SQL operations
-- `file_operations_dag.py` - CSV file processing
-
-**Level 2: Intermediate**
-- `csv_to_postgres.py` - CSV to PostgreSQL pipeline
-- `csv_to_postgres_dq.py` - separate data quality checks for loaded orders
-- `data_processing_dag.py` - ETL pipeline with multiple steps
-- `branching_dag.py` - Conditional task execution
-
-#### 4. Sample Data Structure
-```
-data/
-├── input/
-│   ├── customers.csv
-│   ├── orders.csv
-│   └── products.csv
-├── output/
-└── logs/
-```
-
-#### 5. Learning Progression
-
-**Week 1: Airflow Basics**
-- DAG structure and syntax
-- Basic operators (PythonOperator, BashOperator)
-- Task dependencies
-
-**Week 2: SQL Integration**
-- PostgreSQL connections
-- SQL execution in tasks
-- Data transformation
-
-**Week 3: Real-world Scenarios**
-- Error handling and retries
-- Parameter passing
-- Scheduling
-
-### Implementation Steps
-
-1. **Fix Docker Compose** - Remove duplicate sections, add missing services
-2. **Create .env file** - Hardcode all environment variables
-3. **Setup directories** - Create dags/, data/input/, data/output/
-4. **Create sample DAGs** - From simple to complex
-5. **Add sample data** - CSV files for practical exercises
-6. **Create requirements.txt** - Essential Python packages
-7. **Update documentation** - Clear step-by-step instructions
-8. **Test setup** - Ensure everything works end-to-end
-
-### Key Educational Principles
-
-- **Simplicity First** - Start with minimal configuration
-- **Progressive Complexity** - Build skills step by step
-- **Practical Focus** - Real data processing tasks
-- **Immediate Feedback** - Students see results quickly
-
-### Sample Exercise Structure
-
-Each DAG will include:
-- Clear comments explaining each component
-- Step-by-step task definitions
-- Expected output descriptions
-- Common pitfalls and solutions
-
-### Technical Requirements
-
-- Docker and Docker Compose
-- Basic Python knowledge
-- Basic SQL knowledge
-- Web browser for Airflow UI

From a94d31db065102dfbe17031c51e30fca65bf88b2 Mon Sep 17 00:00:00 2001
From: Dmitry Dementev <me@dementev.space>
Date: Thu, 26 Mar 2026 23:07:24 +0300
Subject: [PATCH 2/3] =?UTF-8?q?docs(tasks):=20=D0=BF=D0=B5=D1=80=D0=B5?=
 =?UTF-8?q?=D1=80=D0=B0=D0=B1=D0=BE=D1=82=D0=B0=D0=BD=D1=8B=20=D1=83=D1=87?=
 =?UTF-8?q?=D0=B5=D0=B1=D0=BD=D1=8B=D0=B5=20=D0=B7=D0=B0=D0=B4=D0=B0=D0=BD?=
 =?UTF-8?q?=D0=B8=D1=8F=20=D0=BF=D0=BE=20=D0=BE=D0=B1=D1=80=D0=B0=D1=82?=
 =?UTF-8?q?=D0=BD=D0=BE=D0=B9=20=D1=81=D0=B2=D1=8F=D0=B7=D0=B8=20=D0=BC?=
 =?UTF-8?q?=D0=B5=D0=BD=D1=82=D0=B8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Зачем:
  - менти жаловался на слабую детализацию заданий, непонятные формулировки
    («ключевые метрики бизнеса», «мониторинг и логирование») и нереалистичное время.
- Что:
  - убрано время выполнения из всех заданий.
  - добавлены лаконичные подсказки: куда смотреть, какой API использовать, подводные камни.
  - переписано задание 2 для data_processing_dag (было «дашборд с HTML» → «расширение текстового отчёта»).
  - переписано задание 3 для error_handling_dag (было абстрактный «мониторинг» → конкретное «logging вместо print»).
- Проверка:
  - вычитка документа, проверка что подсказки не содержат готовых решений.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 airflow-docker/educational-tasks.md | 280 ++++++++++++++++++++--------
 1 file changed, 199 insertions(+), 81 deletions(-)

diff --git a/airflow-docker/educational-tasks.md b/airflow-docker/educational-tasks.md
index 3cbfecc..a40dbad 100644
--- a/airflow-docker/educational-tasks.md
+++ b/airflow-docker/educational-tasks.md
@@ -2,139 +2,182 @@
 
 Этот документ содержит практические задания для закрепления знаний по Apache Airflow. Каждое задание соответствует одному из учебных DAG'ов и направлено на лучшее понимание конкретных концепций.
 
-## 📋 Общие инструкции
+## Общие инструкции
 
 Перед выполнением заданий:
 1. Убедитесь, что стенд Airflow запущен: `docker-compose up -d`
 2. Проверьте доступность интерфейса: http://localhost:8080
 3. Ознакомьтесь с соответствующим DAG'ом в интерфейсе Airflow
+4. Откройте исходный код DAG'а и прочитайте описание в начале файла
 
 ---
 
-## 🚀 Задания для hello_world_dag.py
+## Задания для hello_world_dag.py
 
-**Цель:** Освоить базовые концепции Airflow - создание задач, зависимости, операторы.
+**Цель:** Освоить базовые концепции Airflow — создание задач, зависимости, операторы.
 
 ### Задание 1: Модификация существующих задач
-**Сложность:** 🟢 Начальная
-**Время выполнения:** 10-15 минут
+**Сложность:** Начальная
 
 **Задача:**
 - Откройте файл `hello_world_dag.py`
 - Измените текст в функциях `print_hello()`, `print_date()`, `print_goodbye()` на русский язык
 - Добавьте новую задачу, которая выводит текущую погоду (можно использовать фиктивные данные)
 
+**Подсказка:**
+- Новая задача создаётся по аналогии с `date_task` — нужна Python-функция и `PythonOperator` для неё
+- Не забудьте встроить задачу в цепочку зависимостей (`>>`)
+- После сохранения файла DAG подхватится автоматически — запустите его вручную через UI и проверьте логи
+
 **Цель задания:** Понять структуру DAG, научиться добавлять и модифицировать задачи.
 
 ### Задание 2: Изменение расписания
-**Сложность:** 🟢 Начальная  
-**Время выполнения:** 5-10 минут
+**Сложность:** Начальная
 
 **Задача:**
 - Измените `schedule_interval` с ежедневного на еженедельное выполнение
 - Добавьте параметр `max_active_runs` для ограничения одновременных запусков
 - Проверьте изменения в интерфейсе Airflow
 
+**Подсказка:**
+- Расписание задаётся в параметрах `DAG(...)` — ищите `schedule_interval`
+- Можно использовать `timedelta(weeks=1)` или cron-выражение (например, `'0 0 * * 1'`)
+- `max_active_runs` — тоже параметр `DAG()`
+
 **Цель задания:** Освоить настройку расписания и параметров выполнения DAG.
 
 ### Задание 3: Добавление BashOperator
-**Сложность:** 🟡 Средняя
-**Время выполнения:** 15-20 минут
+**Сложность:** Средняя
 
 **Задача:**
 - Добавьте задачу с `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/<ваш_файл>`
+
 **Цель задания:** Научиться работать с разными типами операторов.
 
 ---
 
-## 🗄️ Задания для sql_basic_dag.py
+## Задания для sql_basic_dag.py
 
 **Цель:** Освоить работу с базами данных через Airflow.
 
 ### Задание 1: Расширение структуры таблицы
-**Сложность:** 🟡 Средняя
-**Время выполнения:** 20-25 минут
+**Сложность:** Средняя
 
 **Задача:**
 - Добавьте в таблицу `students_sample` новые поля: `email`, `phone`, `registration_date`
-- Модифицируйте SQL запросы для работы с новой структурой
+- Модифицируйте SQL-запросы для работы с новой структурой
 - Добавьте задачу для обновления существующих записей
 
+**Подсказка:**
+- Новые колонки добавляются в переменную `create_table_sql` — подберите подходящие типы (VARCHAR, DATE и т.д.)
+- Не забудьте обновить `insert_data_sql` — количество значений в INSERT должно совпадать с количеством колонок
+- Для задачи UPDATE создайте новую SQL-переменную и новый `PostgresOperator` по аналогии с существующими
+- **Подводный камень:** `TRUNCATE TABLE` не изменяет структуру. Чтобы новые колонки появились, замените его на `DROP TABLE IF EXISTS`
+
 **Цель задания:** Научиться работать с миграциями схемы базы данных.
 
 ### Задание 2: Создание отчетов
-**Сложность:** 🟡 Средняя
-**Время выполнения:** 25-30 минут
+**Сложность:** Средняя
 
 **Задача:**
 - Добавьте задачу, которая создает сводный отчет по данным студентов
 - Отчет должен содержать: количество студентов, средний возраст, распределение по возрасту
 - Сохраните отчет в файл в папке `/opt/airflow/data/output/`
 
+**Подсказка:**
+- Здесь понадобится `PythonOperator`, а не `PostgresOperator` — потому что нужно получить результат SQL и записать в файл
+- Для подключения к БД из Python-функции используйте `PostgresHook` (импорт: `from airflow.providers.postgres.hooks.postgres import PostgresHook`)
+- У хука есть метод `get_conn()`, который возвращает обычное psycopg2-соединение — дальше работайте через `cursor.execute()` и `fetchone()`
+- Задачу поставьте перед `drop_table` в цепочке зависимостей
+
 **Цель задания:** Освоить создание аналитических отчетов в DAG'ах.
 
 ### Задание 3: Работа с соединениями
-**Сложность:** 🟡 Средняя
-**Время выполнения:** 15-20 минут
+**Сложность:** Средняя
 
 **Задача:**
 - Создайте новое соединение к базе данных через Airflow UI
 - Модифицируйте DAG для использования нового соединения
 - Добавьте обработку ошибок подключения к БД
 
+**Подсказка:**
+- Соединения создаются в UI: Admin → Connections → "+"
+- Параметры для нашей учебной БД: Host = `postgres-training`, Schema = `training`, Login = `student`, Password = `student`, Port = `5432`
+- В DAG-файле соединение указывается через `postgres_conn_id` — замените значение на ваш новый Conn Id
+- Для проверки ошибок попробуйте указать несуществующее соединение и посмотрите, что покажут логи
+
 **Цель задания:** Научиться управлять соединениями с внешними системами.
 
 ---
 
-## 📁 Задания для file_operations_dag.py
+## Задания для file_operations_dag.py
 
 **Цель:** Освоить работу с файлами и данными в Airflow.
 
 ### Задание 1: Модификация генерации данных
-**Сложность:** 🟡 Средняя
-**Время выполнения:** 20-25 минут
+**Сложность:** Средняя
 
 **Задача:**
 - Добавьте новые поля в генерируемые данные: `department`, `experience_years`, `education_level`
 - Модифицируйте валидацию для проверки новых полей
 - Добавьте фильтрацию данных по определенным критериям (например, опыт > 3 года)
 
+**Подсказка:**
+- Генерация данных происходит в функции `generate_sample_data()` — новые поля добавляются в словарь `data`
+- Для `department` подойдёт `random.choice()` со списком отделов, для `experience_years` — привяжите к возрасту (не может быть больше `age - 18`)
+- Валидация — в функции `read_and_validate_data()`: добавьте `assert` по аналогии с существующими
+- Фильтрацию (pandas `df[df['experience_years'] > 3]`) можно добавить в `transform_data()` и сохранить отдельным файлом
+
 **Цель задания:** Научиться работать с различными типами данных и валидацией.
 
 ### Задание 2: Создание дополнительных отчетов
-**Сложность:** 🟡 Средняя
-**Время выполнения:** 25-30 минут
+**Сложность:** Средняя
 
 **Задача:**
 - Создайте отчет в формате JSON с детальной статистикой по отделам
 - Добавьте визуализацию данных с помощью библиотеки matplotlib (сохранение графика в файл)
 - Создайте сводку по зарплатам в разных возрастных категориях
 
+**Подсказка:**
+- Создайте новую функцию и новый `PythonOperator`, добавьте в цепочку после `summary_task`
+- Для JSON-отчёта: загрузите processed_data.csv через pandas, сгруппируйте (`groupby`) и сохраните через `json.dump()`
+- matplotlib может быть не установлен в контейнере — это нормально, сделайте эту часть необязательной (обработайте `ImportError`)
+- Не забудьте `import pandas as pd` внутри функции (не в шапке файла!)
+
 **Цель задания:** Освоить создание комплексных отчетов и визуализацию.
 
 ### Задание 3: Оптимизация обработки
-**Сложность:** 🟠 Продвинутая
-**Время выполнения:** 30-35 минут
+**Сложность:** Продвинутая
 
 **Задача:**
 - Разделите обработку данных на параллельные задачи для разных отделов
 - Добавьте контроль качества данных (проверка на дубликаты, аномалии)
 - Реализуйте механизм повторной обработки при ошибках
 
+**Подсказка:**
+- Параллельные задачи задаются списком: `read_task >> [task_a, task_b, task_c] >> summary_task`
+- Каждая задача фильтрует DataFrame по своему отделу и сохраняет результат в отдельный файл
+- Проверку дубликатов можно сделать через `df['id'].is_unique`
+- Параметры `retries` и `retry_delay` можно задать как у отдельных задач, так и в `default_args`
+
 **Цель задания:** Научиться оптимизировать и делать обработку данных отказоустойчивой.
 
 ---
 
-## 🐘 Задания для csv_to_postgres.py
+## Задания для csv_to_postgres.py
 
 **Цель:** Освоить паттерны загрузки данных (ETL) из файлов в базу данных и применение XCom.
 
 ### Задание 1: Расширение структуры данных
-**Сложность:** 🟢 Начальная
-**Время выполнения:** 15-20 минут
+**Сложность:** Начальная
 
 **Задача:**
 - Добавьте в генератор CSV новую колонку `status` (например, со случайными значениями 'NEW', 'PROCESSING', 'COMPLETED').
@@ -142,39 +185,51 @@
 - Запустите DAG и проверьте, что данные успешно загрузились с новой колонкой.
 - Убедитесь, что сгенерированный файл появился в каталоге `/opt/airflow/data/output/`.
 
+**Подсказка:**
+- Новая колонка добавляется в трёх местах: генерация (`_generate_csv`), DDL (`_create_table`), загрузка (`_load_csv` — список колонок в COPY)
+- В `_generate_csv()` для случайных значений подойдёт `random.choices(['NEW', 'PROCESSING', 'COMPLETED'], k=rows)`
+- **Подводный камень:** если таблица уже существует со старой схемой, новая колонка не появится. Удалите таблицу вручную или добавьте `DROP TABLE` перед `CREATE TABLE`
+
 **Цель задания:** Понять процесс изменения схемы данных на всех этапах пайплайна.
 
 ### Задание 2: Использование PostgresOperator
-**Сложность:** 🟡 Средняя
-**Время выполнения:** 20-25 минут
+**Сложность:** Средняя
 
 **Задача:**
 - Перепишите задачу `create_orders_table`. Сейчас она использует `PythonOperator` и `PostgresHook` внутри Python-функции.
 - Замените её на использование стандартного `PostgresOperator`, используя соединение `postgres_training`.
 - Убедитесь, что пайплайн продолжает работать корректно.
 
+**Подсказка:**
+- Посмотрите, как устроен `sql_basic_dag.py` — там `PostgresOperator` уже используется, можно взять за образец
+- `PostgresOperator` принимает `postgres_conn_id` и `sql` — SQL можно передать прямо строкой
+- После замены проверьте, не осталось ли ссылок на удалённую функцию. `_get_conn()` всё ещё нужна в `_load_csv()`
+
 **Цель задания:** Научиться использовать специализированные операторы для работы с БД вместо кастомного Python-кода.
 
 ---
 
-## 🛡️ Задания для csv_to_postgres_dq.py
+## Задания для csv_to_postgres_dq.py
 
 **Цель:** Освоить подходы к обеспечению качества данных (Data Quality) в Airflow.
 
 ### Задание 1: Новая проверка качества
-**Сложность:** 🟡 Средняя
-**Время выполнения:** 20-25 минут
+**Сложность:** Средняя
 
 **Задача:**
 - Добавьте новую функцию проверки прямо в `csv_to_postgres_dq.py`, которая будет убеждаться, что все значения в колонке `amount` строго больше нуля.
 - Добавьте вызов этой функции как новую задачу в DAG `csv_to_postgres_dq`.
 - Встройте новую задачу в общую цепочку выполнения (например, перед `data_quality_summary`).
 
+**Подсказка:**
+- Возьмите за образец любую из существующих функций проверки (например, `_check_has_rows`) — паттерн одинаковый: подключиться, выполнить SQL, проверить результат, бросить `ValueError` если не ок
+- SQL для проверки: `SELECT COUNT(*) FROM public.orders WHERE amount <= 0`
+- Не забудьте создать `PythonOperator` и добавить задачу в цепочку
+
 **Цель задания:** Научиться расширять набор проверок качества данных.
 
 ### Задание 2: Управление статусом при ошибках (Trigger Rules)
-**Сложность:** 🟠 Продвинутая
-**Время выполнения:** 25-30 минут
+**Сложность:** Продвинутая
 
 **Задача:**
 - Смоделируйте ошибку (например, временно измените данные так, чтобы проверки не прошли).
@@ -182,134 +237,181 @@
 - Измените параметры задач так (с помощью `trigger_rule`), чтобы выполнялись *все* проверки, даже если некоторые из них упали.
 - Сделайте так, чтобы задача `data_quality_summary` могла анализировать статусы предыдущих задач и отражать общий итог.
 
+**Подсказка:**
+- Для моделирования ошибки: измените `EXPECTED_ORDERS_SCHEMA` так, чтобы схема не совпала. Запустите — увидите, что все задачи после упавшей будут skipped
+- Ключевой параметр — `trigger_rule` у `PythonOperator`. Значение `"all_done"` означает: «запустись в любом случае, когда все upstream завершились (успешно или нет)»
+- Для анализа статусов в `dq_summary`: функция может принимать `**context` и через `context['ti']` получить информацию о статусах предыдущих задач
+- После экспериментов не забудьте вернуть `EXPECTED_ORDERS_SCHEMA` к правильным значениям
+
 **Цель задания:** Освоить продвинутую маршрутизацию статусов задач с помощью `trigger_rule`.
 
 ---
 
-## 🔄 Задания для data_processing_dag.py
+## Задания для data_processing_dag.py
 
 **Цель:** Освоить ETL процессы и работу с бизнес-логикой.
 
 ### Задание 1: Расширение ETL пайплайна
-**Сложность:** 🟡 Средняя
-**Время выполнения:** 30-35 минут
+**Сложность:** Средняя
 
 **Задача:**
-- Добавьте новый источник данных - файл с информацией о продуктах
+- Добавьте новый источник данных — файл с информацией о продуктах
 - Создайте задачу для объединения данных о заказах с информацией о продуктах
 - Добавьте расчет общей выручки по продуктам
 
+**Подсказка:**
+- В `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]`
+
 **Цель задания:** Научиться работать с множественными источниками данных.
 
-### Задание 2: Создание дашборда
-**Сложность:** 🟠 Продвинутая
-**Время выполнения:** 35-40 минут
+### Задание 2: Расширение текстового отчёта
+**Сложность:** Средняя
 
 **Задача:**
-- Создайте HTML-отчет с ключевыми метриками бизнеса
-- Добавьте графики продаж по дням и продуктам
-- Реализуйте отправку отчета по email (симуляция)
+- Расширьте функцию `generate_report()` так, чтобы отчёт содержал больше полезной статистики
+- Добавьте в отчёт: топ-3 самых дорогих заказа, распределение заказов по месяцам, среднюю сумму заказа по каждому клиенту
+- Сохраните расширенный отчёт в `/opt/airflow/data/output/detailed_report.txt`
 
-**Цель задания:** Освоить создание бизнес-отчетов и дашбордов.
+**Подсказка:**
+- Всё делается в существующей функции `generate_report()` — дополните её
+- Полезные методы pandas: `df.nlargest()`, `df.groupby(...).mean()`, `df.groupby(...).sum()`
+- Можно записать результат в тот же файл или создать отдельный
+
+**Цель задания:** Освоить создание аналитических отчётов с помощью pandas.
 
 ### Задание 3: Мониторинг качества данных
-**Сложность:** 🟠 Продвинутая
-**Время выполнения:** 25-30 минут
+**Сложность:** Продвинутая
 
 **Задача:**
 - Добавьте проверки качества данных на каждом этапе ETL
 - Создайте механизм оповещения о проблемах с данными
 - Реализуйте архивирование обработанных данных
 
+**Подсказка:**
+- Создайте отдельную функцию валидации (проверка: строки > 0, нет NULL в ключевых полях, суммы положительные) и поставьте её между extract и transform
+- Для оповещения используйте `on_failure_callback` в `default_args` — это функция, которая вызывается при падении любой задачи. Она получает `context` с информацией об ошибке
+- Для архивирования: скопируйте результат в файл с датой в имени (например, `shutil.copy()` + `datetime.now().strftime(...)`)
+
 **Цель задания:** Научиться обеспечивать качество данных в ETL процессах.
 
 ---
 
-## 🌿 Задания для branching_dag.py
+## Задания для branching_dag.py
 
 **Цель:** Освоить условную логику и ветвление в Airflow.
 
 ### Задание 1: Модификация условий ветвления
-**Сложность:** 🟡 Средняя
-**Время выполнения:** 20-25 минут
+**Сложность:** Средняя
 
 **Задача:**
 - Измените условие в функции `check_data_quality()` на основе реальных критериев (например, размер файла)
 - Добавьте третью ветку обработки для данных "требующих ручной проверки"
 - Настройте разные триггерные правила для слияния веток
 
+**Подсказка:**
+- Сейчас функция выбирает ветку случайно — замените `random.random()` на проверку чего-то реального (например, `os.path.getsize()` для размера файла)
+- Функция `BranchPythonOperator` возвращает `task_id` ветки для выполнения. Для третьей ветки — верните третий `task_id`
+- У `merge_task` проверьте `trigger_rule` — при ветвлении непройденные ветки получают статус `skipped`
+
 **Цель задания:** Научиться создавать сложные условия ветвления.
 
 ### Задание 2: Реализация реального сценария
-**Сложность:** 🟠 Продвинутая
-**Время выполнения:** 30-35 минут
+**Сложность:** Продвинутая
 
 **Задача:**
 - Создайте реальные задачи обработки для CSV и JSON форматов
 - Добавьте валидацию данных в каждой ветке
 - Реализуйте механизм сравнения результатов из разных веток
 
+**Подсказка:**
+- Вместо `print("Обработка CSV...")` загрузите реальный файл (например, `sample_data.csv`) и посчитайте статистику
+- Результат каждой ветки сохраните в отдельный файл — в `merge_results()` загрузите оба (если существуют) и сравните
+- Учтите, что при ветвлении выполняется только одна ветка — функция слияния должна обрабатывать случай, когда один из файлов отсутствует
+
 **Цель задания:** Применить ветвление в реальном сценарии обработки данных.
 
 ### Задание 3: Динамическое ветвление
-**Сложность:** 🟠 Продвинутая
-**Время выполнения:** 25-30 минут
+**Сложность:** Продвинутая
 
 **Задача:**
 - Реализуйте ветвление на основе внешних параметров (например, переданных через Variables)
 - Добавьте обработку случая, когда ни одна ветка не подходит
 - Создайте механизм логирования выбранного пути выполнения
 
+**Подсказка:**
+- Variables создаются в UI: Admin → Variables. Для чтения в коде: `Variable.get('key', default_var='значение')`
+- Для случая «ни одна ветка не подходит» добавьте `DummyOperator` как ветку-заглушку
+- `BranchPythonOperator` должен всегда возвращать существующий `task_id` — иначе DAG упадёт с ошибкой
+
 **Цель задания:** Освоить динамическое принятие решений в DAG'ах.
 
 ---
 
-## ⚠️ Задания для error_handling_dag.py
+## Задания для error_handling_dag.py
 
 **Цель:** Освоить обработку ошибок и создание отказоустойчивых пайплайнов.
 
 ### Задание 1: Настройка стратегий повторения
-**Сложность:** 🟡 Средняя
-**Время выполнения:** 20-25 минут
+**Сложность:** Средняя
 
 **Задача:**
 - Измените параметры `retries` и `retry_delay` для разных задач
 - Добавьте экспоненциальную задержку между повторными попытками
 - Реализуйте кастомный обработчик ошибок для конкретных исключений
 
+**Подсказка:**
+- Параметры повторения можно задать на уровне задачи (перекроют `default_args`): `retries`, `retry_delay`, `retry_exponential_backoff`, `max_retry_delay`
+- Для кастомного обработчика: параметр `on_retry_callback` принимает функцию с аргументом `context`, из которого можно получить номер попытки через `context['ti'].try_number`
+- Запустите DAG несколько раз и посмотрите в логах, как меняется задержка между попытками
+
 **Цель задания:** Научиться настраивать стратегии обработки ошибок.
 
 ### Задание 2: Создание комплексной обработки ошибок
-**Сложность:** 🟠 Продвинутая
-**Время выполнения:** 30-35 минут
+**Сложность:** Продвинутая
 
 **Задача:**
 - Добавьте задачи для разных типов ошибок (сетевая ошибка, ошибка данных, системная ошибка)
 - Создайте механизм эскалации ошибок (после N неудачных попыток)
 - Реализуйте отправку уведомлений о критических ошибках
 
+**Подсказка:**
+- Создайте функции по аналогии с `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:
+          # сюда — логику эскалации
+  ```
+
 **Цель задания:** Освоить создание комплексной системы обработки ошибок.
 
-### Задание 3: Мониторинг и логирование
-**Сложность:** 🟠 Продвинутая
-**Время выполнения:** 25-30 минут
+### Задание 3: Добавление логирования через модуль logging
+**Сложность:** Средняя
 
 **Задача:**
-- Добавьте детальное логирование всех этапов выполнения
-- Создайте задачу для анализа логов и генерации отчетов об ошибках
-- Реализуйте механизм автоматического восстановления после сбоев
+- Замените все `print()` в функциях DAG'а на вызовы стандартного модуля `logging`
+- Добавьте в каждую функцию логирование начала и окончания выполнения
+- Посмотрите, как логи отображаются в интерфейсе Airflow (вкладка Log у каждой задачи)
 
-**Цель задания:** Научиться создавать системы мониторинга и отладки.
+**Подсказка:**
+- Стандартный паттерн: `import logging` в начале файла, затем `log = logging.getLogger(__name__)` в функции
+- Уровни: `log.info()` для штатных событий, `log.warning()` для предупреждений, `log.error()` для ошибок
+- В UI Airflow откройте выполненную задачу → вкладка **Log** — сообщения `logging` отображаются с метками уровня и временем, в отличие от `print()`
+
+**Цель задания:** Научиться использовать стандартное логирование Python в задачах Airflow и читать логи через UI.
 
 ---
 
-## 🌟 Дополнительные задания: пулы, XCom, TaskGroup и алертинг
+## Дополнительные задания: пулы, XCom, TaskGroup и алертинг
 
 **Цель:** Освоить продвинутые возможности оркестрации — управление ресурсами, обмен данными между задачами, группировку и оповещения.
 
 ### Задание 1: Пулы и управление ресурсами
-**Сложность:** 🟡 Средняя  
-**Время выполнения:** 20-25 минут
+**Сложность:** Средняя
 
 **Задача:**
 - В интерфейсе Airflow создайте пул `backup_pool` с 2 слотами
@@ -317,22 +419,31 @@
 - Назначьте этим задачам параметр `pool="backup_pool"` и настройте `pool_slots` так, чтобы одна из задач занимала 2 слота, а другая — 1
 - Наблюдайте в UI, что одновременно запускается не более 2 задач из этого пула
 
+**Подсказка:**
+- Пул создаётся в UI: Admin → Pools → "+"
+- У `PythonOperator` есть параметры `pool` (имя пула) и `pool_slots` (сколько слотов занимает задача)
+- Если задача занимает 2 слота из 2 доступных — вторая задача будет ждать, даже если они не связаны зависимостями
+
 **Цель задания:** Научиться управлять параллелизмом задач через пулы и `pool_slots`.
 
 ### Задание 2: Обмен данными через XCom
-**Сложность:** 🟡 Средняя  
-**Время выполнения:** 20-25 минут
+**Сложность:** Средняя
 
 **Задача:**
 - В том же DAG добавьте задачу `calculate_metrics` (PythonOperator), которая возвращает словарь с агрегированными показателями, например: `{"total_orders": ..., "avg_amount": ...}`
 - Добавьте задачу `log_metrics`, которая с помощью `xcom_pull` читает результат `calculate_metrics` и выводит значения в лог
 - Для одной из задач продемонстрируйте использование XCom в Jinja-шаблоне (например, в `bash_command` или SQL-запросе)
 
+**Подсказка:**
+- Любое значение, которое функция возвращает через `return`, автоматически сохраняется в XCom
+- Для чтения XCom в другой задаче: функция принимает `**context`, затем `context['ti'].xcom_pull(task_ids='имя_задачи')`
+- В Jinja-шаблонах (параметры `bash_command`, `sql` и др.) XCom доступен через `{{ ti.xcom_pull(task_ids='...') }}`
+- Посмотреть сохранённые XCom-значения можно в UI: откройте задачу → вкладка **XCom**
+
 **Цель задания:** Освоить передачу результатов между задачами через XCom и их использование в шаблонах.
 
 ### Задание 3: TaskGroup и алертинг
-**Сложность:** 🟠 Продвинутая  
-**Время выполнения:** 30-35 минут
+**Сложность:** Продвинутая
 
 **Задача:**
 - Объедините логически связанные задачи (например, `extract` / `transform` / `load`) в `TaskGroup`'ы
@@ -340,21 +451,28 @@
 - Настройте для задачи уведомления `trigger_rule=TriggerRule.ALL_DONE`, чтобы уведомление отправлялось даже при частичных ошибках
 - При желании вынесите функцию отправки письма в отдельный модуль `utils.py` и импортируйте её в DAG
 
+**Подсказка:**
+- TaskGroup — контекстный менеджер: `with TaskGroup("имя") as group:` — внутри определяете задачи как обычно
+- Зависимости работают на уровне групп: `group_a >> group_b`
+- `TriggerRule.ALL_DONE` (из `airflow.utils.trigger_rule`) означает: «запустись, когда все upstream завершились — неважно, с успехом или ошибкой»
+- В Graph View группы отображаются как складные блоки — удобно для больших DAG'ов
+
 **Цель задания:** Научиться группировать задачи с помощью TaskGroup и строить схему оповещений о статусе пайплайна.
 
 ---
 
-## 🎯 Рекомендации по выполнению
+## Рекомендации по выполнению
 
 1. **Начинайте с простых заданий** и постепенно переходите к сложным
 2. **Тестируйте каждое изменение** через интерфейс Airflow
-3. **Изучайте логи выполнения** для понимания поведения задач
+3. **Изучайте логи выполнения** для понимания поведения задач (вкладка Log у каждой задачи в UI)
 4. **Экспериментируйте** с разными настройками и параметрами
-5. **Документируйте** свои решения и находки
+5. **Не бойтесь ломать** — DAG'и можно откатить через `git checkout`, а таблицы пересоздать
 
-## 📊 Оценка прогресса
+## Оценка прогресса
 
-- 🟢 **Начальный уровень:** Выполнены задания для 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 и алертингу
-Удачи в изучении Apache Airflow! 🚀
+- **Начальный уровень:** Выполнены задания для 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 и алертингу
+
+Удачи в изучении Apache Airflow!

From 50207811a8bf03e56d91de7756851727fbc3b312 Mon Sep 17 00:00:00 2001
From: Dmitry Dementev <me@dementev.space>
Date: Fri, 27 Mar 2026 21:49:22 +0300
Subject: [PATCH 3/3] =?UTF-8?q?chore(repo):=20=D0=B4=D0=BE=D0=B1=D0=B0?=
 =?UTF-8?q?=D0=B2=D0=BB=D0=B5=D0=BD=20CLAUDE.md=20=D1=81=D0=BE=20=D1=81?=
 =?UTF-8?q?=D1=81=D1=8B=D0=BB=D0=BA=D0=BE=D0=B9=20=D0=BD=D0=B0=20AGENTS.md?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Зачем:
  - чтобы инструменты и агенты могли находить единый файл с инструкциями для репозитория.
- Что:
  - добавлен файл CLAUDE.md.
  - в файл помещена ссылка на AGENTS.md как точку входа в актуальные правила.
- Проверка:
  - git status --short.
---
 CLAUDE.md | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 CLAUDE.md

diff --git a/CLAUDE.md b/CLAUDE.md
new file mode 100644
index 0000000..eef4bd2
--- /dev/null
+++ b/CLAUDE.md
@@ -0,0 +1 @@
+@AGENTS.md
\ No newline at end of file