diff --git a/airflow-docker/README.md b/airflow-docker/README.md index 819fa30..f312c46 100644 --- a/airflow-docker/README.md +++ b/airflow-docker/README.md @@ -96,9 +96,17 @@ airflow-docker/ │ ├── hello_world_dag.py # Базовый пример │ ├── sql_basic_dag.py # Работа с SQL │ ├── file_operations_dag.py # Обработка файлов -│ ├── data_processing_dag.py # ETL пайплайн +│ ├── 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 # Обработка ошибок +├── helpers/ # Вспомогательные скрипты +│ └── postgres.py # Функции для работы с БД и DQ +├── tests/ # Unit-тесты для хелперов +│ └── test_postgres_helpers.py # Тестирование DQ проверок +├── sql/ # SQL скрипты и DDL +│ └── base/ # Базовые DDL таблиц ├── data/ # Данные для упражнений │ ├── input/ # Входные данные │ └── output/ # Результаты обработки @@ -129,6 +137,8 @@ airflow-docker/ **Примеры DAG:** - `file_operations_dag.py` - работа с файлами +- `csv_to_postgres.py` - загрузка данных из CSV в PostgreSQL +- `csv_to_postgres_dq.py` - автоматизированные проверки качества (Data Quality) - `data_processing_dag.py` - ETL процессы ### Продвинутые возможности @@ -159,8 +169,9 @@ airflow-docker/ 1. **Начните с `hello_world_dag.py`** - освоите основы Airflow 2. **Перейдите к `sql_basic_dag.py`** - изучите работу с базами данных 3. **Попрактикуйтесь на `file_operations_dag.py`** - работа с файлами -4. **Освойте ETL на `data_processing_dag.py`** - обработка данных -5. **Изучите продвинутые темы** - ветвление и обработка ошибок +4. **Освойте ETL и DQ на `csv_to_postgres.py` и `csv_to_postgres_dq.py`** - загрузка и валидация данных +5. **Разберите сложный ETL на `data_processing_dag.py`** - обработка данных +6. **Изучите продвинутые темы** - ветвление и обработка ошибок Каждое задание содержит: - Цель и сложность выполнения diff --git a/airflow-docker/educational-tasks.md b/airflow-docker/educational-tasks.md index b1a33e9..e8a8564 100644 --- a/airflow-docker/educational-tasks.md +++ b/airflow-docker/educational-tasks.md @@ -128,6 +128,63 @@ --- +## 🐘 Задания для csv_to_postgres.py + +**Цель:** Освоить паттерны загрузки данных (ETL) из файлов в базу данных и применение XCom. + +### Задание 1: Расширение структуры данных +**Сложность:** 🟢 Начальная +**Время выполнения:** 15-20 минут + +**Задача:** +- Добавьте в генератор CSV новую колонку `status` (например, со случайными значениями 'NEW', 'PROCESSING', 'COMPLETED'). +- Обновите функцию `_create_table`, чтобы учесть новую колонку. +- Запустите DAG и проверьте, что данные успешно загрузились с новой колонкой. + +**Цель задания:** Понять процесс изменения схемы данных на всех этапах пайплайна. + +### Задание 2: Использование PostgresOperator +**Сложность:** 🟡 Средняя +**Время выполнения:** 20-25 минут + +**Задача:** +- Перепишите задачу `create_orders_table`. Сейчас она использует `PythonOperator` и прямое подключение через `psycopg2`. +- Замените её на использование стандартного `PostgresOperator`, используя заранее созданный Connection. +- Убедитесь, что пайплайн продолжает работать корректно. + +**Цель задания:** Научиться использовать специализированные операторы для работы с БД вместо кастомного Python-кода. + +--- + +## 🛡️ Задания для csv_to_postgres_dq.py + +**Цель:** Освоить подходы к обеспечению качества данных (Data Quality) в Airflow. + +### Задание 1: Новая проверка качества +**Сложность:** 🟡 Средняя +**Время выполнения:** 20-25 минут + +**Задача:** +- Добавьте новую функцию проверки в `helpers/postgres.py`, которая будет убеждаться, что все значения в колонке `amount` строго больше нуля. +- Добавьте вызов этой функции как новую задачу в DAG `csv_to_postgres_dq`. +- Встройте новую задачу в общую цепочку выполнения (например, перед `dq_summary`). + +**Цель задания:** Научиться расширять набор проверок качества данных. + +### Задание 2: Управление статусом при ошибках (Trigger Rules) +**Сложность:** 🟠 Продвинутая +**Время выполнения:** 25-30 минут + +**Задача:** +- Смоделируйте ошибку (например, временно измените данные так, чтобы проверки не прошли). +- По умолчанию, если падает одна проверка, следующие не выполняются (поведение `all_success`). +- Измените параметры задач так (с помощью `trigger_rule`), чтобы выполнялись *все* проверки, даже если некоторые из них упали. +- Сделайте так, чтобы задача `dq_summary` могла анализировать статусы предыдущих задач и отражать общий итог. + +**Цель задания:** Освоить продвинутую маршрутизацию статусов задач с помощью `trigger_rule`. + +--- + ## 🔄 Задания для data_processing_dag.py **Цель:** Освоить ETL процессы и работу с бизнес-логикой. @@ -297,7 +354,6 @@ ## 📊 Оценка прогресса - 🟢 **Начальный уровень:** Выполнены задания для hello_world_dag и sql_basic_dag -- 🟡 **Средний уровень:** Выполнены задания для file_operations_dag и data_processing_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! 🚀