docs(airflow): обновлены практические задания
- Зачем: - упражнения должны быть понятнее и соответствовать ожидаемому уровню сложности. - Что: - добавлены подсказки и описания подводных камней. - упрощены задания по отчетам, мониторингу и логированию. - Проверка: - git diff --check.
This commit is contained in:
@@ -2,139 +2,182 @@
|
|||||||
|
|
||||||
Этот документ содержит практические задания для закрепления знаний по Apache Airflow. Каждое задание соответствует одному из учебных DAG'ов и направлено на лучшее понимание конкретных концепций.
|
Этот документ содержит практические задания для закрепления знаний по Apache Airflow. Каждое задание соответствует одному из учебных DAG'ов и направлено на лучшее понимание конкретных концепций.
|
||||||
|
|
||||||
## 📋 Общие инструкции
|
## Общие инструкции
|
||||||
|
|
||||||
Перед выполнением заданий:
|
Перед выполнением заданий:
|
||||||
1. Убедитесь, что стенд Airflow запущен: `docker-compose up -d`
|
1. Убедитесь, что стенд Airflow запущен: `docker-compose up -d`
|
||||||
2. Проверьте доступность интерфейса: http://localhost:8080
|
2. Проверьте доступность интерфейса: http://localhost:8080
|
||||||
3. Ознакомьтесь с соответствующим DAG'ом в интерфейсе Airflow
|
3. Ознакомьтесь с соответствующим DAG'ом в интерфейсе Airflow
|
||||||
|
4. Откройте исходный код DAG'а и прочитайте описание в начале файла
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## 🚀 Задания для hello_world_dag.py
|
## Задания для hello_world_dag.py
|
||||||
|
|
||||||
**Цель:** Освоить базовые концепции Airflow - создание задач, зависимости, операторы.
|
**Цель:** Освоить базовые концепции Airflow — создание задач, зависимости, операторы.
|
||||||
|
|
||||||
### Задание 1: Модификация существующих задач
|
### Задание 1: Модификация существующих задач
|
||||||
**Сложность:** 🟢 Начальная
|
**Сложность:** Начальная
|
||||||
**Время выполнения:** 10-15 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Откройте файл `hello_world_dag.py`
|
- Откройте файл `hello_world_dag.py`
|
||||||
- Измените текст в функциях `print_hello()`, `print_date()`, `print_goodbye()` на русский язык
|
- Измените текст в функциях `print_hello()`, `print_date()`, `print_goodbye()` на русский язык
|
||||||
- Добавьте новую задачу, которая выводит текущую погоду (можно использовать фиктивные данные)
|
- Добавьте новую задачу, которая выводит текущую погоду (можно использовать фиктивные данные)
|
||||||
|
|
||||||
|
**Подсказка:**
|
||||||
|
- Новая задача создаётся по аналогии с `date_task` — нужна Python-функция и `PythonOperator` для неё
|
||||||
|
- Не забудьте встроить задачу в цепочку зависимостей (`>>`)
|
||||||
|
- После сохранения файла DAG подхватится автоматически — запустите его вручную через UI и проверьте логи
|
||||||
|
|
||||||
**Цель задания:** Понять структуру DAG, научиться добавлять и модифицировать задачи.
|
**Цель задания:** Понять структуру DAG, научиться добавлять и модифицировать задачи.
|
||||||
|
|
||||||
### Задание 2: Изменение расписания
|
### Задание 2: Изменение расписания
|
||||||
**Сложность:** 🟢 Начальная
|
**Сложность:** Начальная
|
||||||
**Время выполнения:** 5-10 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Измените `schedule_interval` с ежедневного на еженедельное выполнение
|
- Измените `schedule_interval` с ежедневного на еженедельное выполнение
|
||||||
- Добавьте параметр `max_active_runs` для ограничения одновременных запусков
|
- Добавьте параметр `max_active_runs` для ограничения одновременных запусков
|
||||||
- Проверьте изменения в интерфейсе Airflow
|
- Проверьте изменения в интерфейсе Airflow
|
||||||
|
|
||||||
|
**Подсказка:**
|
||||||
|
- Расписание задаётся в параметрах `DAG(...)` — ищите `schedule_interval`
|
||||||
|
- Можно использовать `timedelta(weeks=1)` или cron-выражение (например, `'0 0 * * 1'`)
|
||||||
|
- `max_active_runs` — тоже параметр `DAG()`
|
||||||
|
|
||||||
**Цель задания:** Освоить настройку расписания и параметров выполнения DAG.
|
**Цель задания:** Освоить настройку расписания и параметров выполнения DAG.
|
||||||
|
|
||||||
### Задание 3: Добавление BashOperator
|
### Задание 3: Добавление BashOperator
|
||||||
**Сложность:** 🟡 Средняя
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 15-20 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Добавьте задачу с `BashOperator`, которая создает текстовый файл в папке `/opt/airflow/data/output/`
|
- Добавьте задачу с `BashOperator`, которая создает текстовый файл в папке `/opt/airflow/data/output/`
|
||||||
- Настройте зависимости так, чтобы эта задача выполнялась между `date_task` и `end_task`
|
- Настройте зависимости так, чтобы эта задача выполнялась между `date_task` и `end_task`
|
||||||
- Убедитесь, что файл создается при каждом запуске DAG
|
- Убедитесь, что файл создается при каждом запуске 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.
|
**Цель:** Освоить работу с базами данных через Airflow.
|
||||||
|
|
||||||
### Задание 1: Расширение структуры таблицы
|
### Задание 1: Расширение структуры таблицы
|
||||||
**Сложность:** 🟡 Средняя
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 20-25 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Добавьте в таблицу `students_sample` новые поля: `email`, `phone`, `registration_date`
|
- Добавьте в таблицу `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: Создание отчетов
|
### Задание 2: Создание отчетов
|
||||||
**Сложность:** 🟡 Средняя
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 25-30 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Добавьте задачу, которая создает сводный отчет по данным студентов
|
- Добавьте задачу, которая создает сводный отчет по данным студентов
|
||||||
- Отчет должен содержать: количество студентов, средний возраст, распределение по возрасту
|
- Отчет должен содержать: количество студентов, средний возраст, распределение по возрасту
|
||||||
- Сохраните отчет в файл в папке `/opt/airflow/data/output/`
|
- Сохраните отчет в файл в папке `/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'ах.
|
**Цель задания:** Освоить создание аналитических отчетов в DAG'ах.
|
||||||
|
|
||||||
### Задание 3: Работа с соединениями
|
### Задание 3: Работа с соединениями
|
||||||
**Сложность:** 🟡 Средняя
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 15-20 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Создайте новое соединение к базе данных через Airflow UI
|
- Создайте новое соединение к базе данных через Airflow UI
|
||||||
- Модифицируйте DAG для использования нового соединения
|
- Модифицируйте 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.
|
**Цель:** Освоить работу с файлами и данными в Airflow.
|
||||||
|
|
||||||
### Задание 1: Модификация генерации данных
|
### Задание 1: Модификация генерации данных
|
||||||
**Сложность:** 🟡 Средняя
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 20-25 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Добавьте новые поля в генерируемые данные: `department`, `experience_years`, `education_level`
|
- Добавьте новые поля в генерируемые данные: `department`, `experience_years`, `education_level`
|
||||||
- Модифицируйте валидацию для проверки новых полей
|
- Модифицируйте валидацию для проверки новых полей
|
||||||
- Добавьте фильтрацию данных по определенным критериям (например, опыт > 3 года)
|
- Добавьте фильтрацию данных по определенным критериям (например, опыт > 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: Создание дополнительных отчетов
|
### Задание 2: Создание дополнительных отчетов
|
||||||
**Сложность:** 🟡 Средняя
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 25-30 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Создайте отчет в формате JSON с детальной статистикой по отделам
|
- Создайте отчет в формате JSON с детальной статистикой по отделам
|
||||||
- Добавьте визуализацию данных с помощью библиотеки matplotlib (сохранение графика в файл)
|
- Добавьте визуализацию данных с помощью библиотеки matplotlib (сохранение графика в файл)
|
||||||
- Создайте сводку по зарплатам в разных возрастных категориях
|
- Создайте сводку по зарплатам в разных возрастных категориях
|
||||||
|
|
||||||
|
**Подсказка:**
|
||||||
|
- Создайте новую функцию и новый `PythonOperator`, добавьте в цепочку после `summary_task`
|
||||||
|
- Для JSON-отчёта: загрузите processed_data.csv через pandas, сгруппируйте (`groupby`) и сохраните через `json.dump()`
|
||||||
|
- matplotlib может быть не установлен в контейнере — это нормально, сделайте эту часть необязательной (обработайте `ImportError`)
|
||||||
|
- Не забудьте `import pandas as pd` внутри функции (не в шапке файла!)
|
||||||
|
|
||||||
**Цель задания:** Освоить создание комплексных отчетов и визуализацию.
|
**Цель задания:** Освоить создание комплексных отчетов и визуализацию.
|
||||||
|
|
||||||
### Задание 3: Оптимизация обработки
|
### Задание 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.
|
**Цель:** Освоить паттерны загрузки данных (ETL) из файлов в базу данных и применение XCom.
|
||||||
|
|
||||||
### Задание 1: Расширение структуры данных
|
### Задание 1: Расширение структуры данных
|
||||||
**Сложность:** 🟢 Начальная
|
**Сложность:** Начальная
|
||||||
**Время выполнения:** 15-20 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Добавьте в генератор CSV новую колонку `status` (например, со случайными значениями 'NEW', 'PROCESSING', 'COMPLETED').
|
- Добавьте в генератор CSV новую колонку `status` (например, со случайными значениями 'NEW', 'PROCESSING', 'COMPLETED').
|
||||||
@@ -142,39 +185,51 @@
|
|||||||
- Запустите DAG и проверьте, что данные успешно загрузились с новой колонкой.
|
- Запустите DAG и проверьте, что данные успешно загрузились с новой колонкой.
|
||||||
- Убедитесь, что сгенерированный файл появился в каталоге `/opt/airflow/data/output/`.
|
- Убедитесь, что сгенерированный файл появился в каталоге `/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
|
### Задание 2: Использование PostgresOperator
|
||||||
**Сложность:** 🟡 Средняя
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 20-25 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Перепишите задачу `create_orders_table`. Сейчас она использует `PythonOperator` и `PostgresHook` внутри Python-функции.
|
- Перепишите задачу `create_orders_table`. Сейчас она использует `PythonOperator` и `PostgresHook` внутри Python-функции.
|
||||||
- Замените её на использование стандартного `PostgresOperator`, используя соединение `postgres_training`.
|
- Замените её на использование стандартного `PostgresOperator`, используя соединение `postgres_training`.
|
||||||
- Убедитесь, что пайплайн продолжает работать корректно.
|
- Убедитесь, что пайплайн продолжает работать корректно.
|
||||||
|
|
||||||
|
**Подсказка:**
|
||||||
|
- Посмотрите, как устроен `sql_basic_dag.py` — там `PostgresOperator` уже используется, можно взять за образец
|
||||||
|
- `PostgresOperator` принимает `postgres_conn_id` и `sql` — SQL можно передать прямо строкой
|
||||||
|
- После замены проверьте, не осталось ли ссылок на удалённую функцию. `_get_conn()` всё ещё нужна в `_load_csv()`
|
||||||
|
|
||||||
**Цель задания:** Научиться использовать специализированные операторы для работы с БД вместо кастомного Python-кода.
|
**Цель задания:** Научиться использовать специализированные операторы для работы с БД вместо кастомного Python-кода.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## 🛡️ Задания для csv_to_postgres_dq.py
|
## Задания для csv_to_postgres_dq.py
|
||||||
|
|
||||||
**Цель:** Освоить подходы к обеспечению качества данных (Data Quality) в Airflow.
|
**Цель:** Освоить подходы к обеспечению качества данных (Data Quality) в Airflow.
|
||||||
|
|
||||||
### Задание 1: Новая проверка качества
|
### Задание 1: Новая проверка качества
|
||||||
**Сложность:** 🟡 Средняя
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 20-25 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Добавьте новую функцию проверки прямо в `csv_to_postgres_dq.py`, которая будет убеждаться, что все значения в колонке `amount` строго больше нуля.
|
- Добавьте новую функцию проверки прямо в `csv_to_postgres_dq.py`, которая будет убеждаться, что все значения в колонке `amount` строго больше нуля.
|
||||||
- Добавьте вызов этой функции как новую задачу в DAG `csv_to_postgres_dq`.
|
- Добавьте вызов этой функции как новую задачу в DAG `csv_to_postgres_dq`.
|
||||||
- Встройте новую задачу в общую цепочку выполнения (например, перед `data_quality_summary`).
|
- Встройте новую задачу в общую цепочку выполнения (например, перед `data_quality_summary`).
|
||||||
|
|
||||||
|
**Подсказка:**
|
||||||
|
- Возьмите за образец любую из существующих функций проверки (например, `_check_has_rows`) — паттерн одинаковый: подключиться, выполнить SQL, проверить результат, бросить `ValueError` если не ок
|
||||||
|
- SQL для проверки: `SELECT COUNT(*) FROM public.orders WHERE amount <= 0`
|
||||||
|
- Не забудьте создать `PythonOperator` и добавить задачу в цепочку
|
||||||
|
|
||||||
**Цель задания:** Научиться расширять набор проверок качества данных.
|
**Цель задания:** Научиться расширять набор проверок качества данных.
|
||||||
|
|
||||||
### Задание 2: Управление статусом при ошибках (Trigger Rules)
|
### Задание 2: Управление статусом при ошибках (Trigger Rules)
|
||||||
**Сложность:** 🟠 Продвинутая
|
**Сложность:** Продвинутая
|
||||||
**Время выполнения:** 25-30 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Смоделируйте ошибку (например, временно измените данные так, чтобы проверки не прошли).
|
- Смоделируйте ошибку (например, временно измените данные так, чтобы проверки не прошли).
|
||||||
@@ -182,134 +237,181 @@
|
|||||||
- Измените параметры задач так (с помощью `trigger_rule`), чтобы выполнялись *все* проверки, даже если некоторые из них упали.
|
- Измените параметры задач так (с помощью `trigger_rule`), чтобы выполнялись *все* проверки, даже если некоторые из них упали.
|
||||||
- Сделайте так, чтобы задача `data_quality_summary` могла анализировать статусы предыдущих задач и отражать общий итог.
|
- Сделайте так, чтобы задача `data_quality_summary` могла анализировать статусы предыдущих задач и отражать общий итог.
|
||||||
|
|
||||||
|
**Подсказка:**
|
||||||
|
- Для моделирования ошибки: измените `EXPECTED_ORDERS_SCHEMA` так, чтобы схема не совпала. Запустите — увидите, что все задачи после упавшей будут skipped
|
||||||
|
- Ключевой параметр — `trigger_rule` у `PythonOperator`. Значение `"all_done"` означает: «запустись в любом случае, когда все upstream завершились (успешно или нет)»
|
||||||
|
- Для анализа статусов в `dq_summary`: функция может принимать `**context` и через `context['ti']` получить информацию о статусах предыдущих задач
|
||||||
|
- После экспериментов не забудьте вернуть `EXPECTED_ORDERS_SCHEMA` к правильным значениям
|
||||||
|
|
||||||
**Цель задания:** Освоить продвинутую маршрутизацию статусов задач с помощью `trigger_rule`.
|
**Цель задания:** Освоить продвинутую маршрутизацию статусов задач с помощью `trigger_rule`.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## 🔄 Задания для data_processing_dag.py
|
## Задания для data_processing_dag.py
|
||||||
|
|
||||||
**Цель:** Освоить ETL процессы и работу с бизнес-логикой.
|
**Цель:** Освоить ETL процессы и работу с бизнес-логикой.
|
||||||
|
|
||||||
### Задание 1: Расширение 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: Создание дашборда
|
### Задание 2: Расширение текстового отчёта
|
||||||
**Сложность:** 🟠 Продвинутая
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 35-40 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Создайте HTML-отчет с ключевыми метриками бизнеса
|
- Расширьте функцию `generate_report()` так, чтобы отчёт содержал больше полезной статистики
|
||||||
- Добавьте графики продаж по дням и продуктам
|
- Добавьте в отчёт: топ-3 самых дорогих заказа, распределение заказов по месяцам, среднюю сумму заказа по каждому клиенту
|
||||||
- Реализуйте отправку отчета по email (симуляция)
|
- Сохраните расширенный отчёт в `/opt/airflow/data/output/detailed_report.txt`
|
||||||
|
|
||||||
**Цель задания:** Освоить создание бизнес-отчетов и дашбордов.
|
**Подсказка:**
|
||||||
|
- Всё делается в существующей функции `generate_report()` — дополните её
|
||||||
|
- Полезные методы pandas: `df.nlargest()`, `df.groupby(...).mean()`, `df.groupby(...).sum()`
|
||||||
|
- Можно записать результат в тот же файл или создать отдельный
|
||||||
|
|
||||||
|
**Цель задания:** Освоить создание аналитических отчётов с помощью pandas.
|
||||||
|
|
||||||
### Задание 3: Мониторинг качества данных
|
### Задание 3: Мониторинг качества данных
|
||||||
**Сложность:** 🟠 Продвинутая
|
**Сложность:** Продвинутая
|
||||||
**Время выполнения:** 25-30 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Добавьте проверки качества данных на каждом этапе ETL
|
- Добавьте проверки качества данных на каждом этапе ETL
|
||||||
- Создайте механизм оповещения о проблемах с данными
|
- Создайте механизм оповещения о проблемах с данными
|
||||||
- Реализуйте архивирование обработанных данных
|
- Реализуйте архивирование обработанных данных
|
||||||
|
|
||||||
|
**Подсказка:**
|
||||||
|
- Создайте отдельную функцию валидации (проверка: строки > 0, нет NULL в ключевых полях, суммы положительные) и поставьте её между extract и transform
|
||||||
|
- Для оповещения используйте `on_failure_callback` в `default_args` — это функция, которая вызывается при падении любой задачи. Она получает `context` с информацией об ошибке
|
||||||
|
- Для архивирования: скопируйте результат в файл с датой в имени (например, `shutil.copy()` + `datetime.now().strftime(...)`)
|
||||||
|
|
||||||
**Цель задания:** Научиться обеспечивать качество данных в ETL процессах.
|
**Цель задания:** Научиться обеспечивать качество данных в ETL процессах.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## 🌿 Задания для branching_dag.py
|
## Задания для branching_dag.py
|
||||||
|
|
||||||
**Цель:** Освоить условную логику и ветвление в Airflow.
|
**Цель:** Освоить условную логику и ветвление в Airflow.
|
||||||
|
|
||||||
### Задание 1: Модификация условий ветвления
|
### Задание 1: Модификация условий ветвления
|
||||||
**Сложность:** 🟡 Средняя
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 20-25 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Измените условие в функции `check_data_quality()` на основе реальных критериев (например, размер файла)
|
- Измените условие в функции `check_data_quality()` на основе реальных критериев (например, размер файла)
|
||||||
- Добавьте третью ветку обработки для данных "требующих ручной проверки"
|
- Добавьте третью ветку обработки для данных "требующих ручной проверки"
|
||||||
- Настройте разные триггерные правила для слияния веток
|
- Настройте разные триггерные правила для слияния веток
|
||||||
|
|
||||||
|
**Подсказка:**
|
||||||
|
- Сейчас функция выбирает ветку случайно — замените `random.random()` на проверку чего-то реального (например, `os.path.getsize()` для размера файла)
|
||||||
|
- Функция `BranchPythonOperator` возвращает `task_id` ветки для выполнения. Для третьей ветки — верните третий `task_id`
|
||||||
|
- У `merge_task` проверьте `trigger_rule` — при ветвлении непройденные ветки получают статус `skipped`
|
||||||
|
|
||||||
**Цель задания:** Научиться создавать сложные условия ветвления.
|
**Цель задания:** Научиться создавать сложные условия ветвления.
|
||||||
|
|
||||||
### Задание 2: Реализация реального сценария
|
### Задание 2: Реализация реального сценария
|
||||||
**Сложность:** 🟠 Продвинутая
|
**Сложность:** Продвинутая
|
||||||
**Время выполнения:** 30-35 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Создайте реальные задачи обработки для CSV и JSON форматов
|
- Создайте реальные задачи обработки для CSV и JSON форматов
|
||||||
- Добавьте валидацию данных в каждой ветке
|
- Добавьте валидацию данных в каждой ветке
|
||||||
- Реализуйте механизм сравнения результатов из разных веток
|
- Реализуйте механизм сравнения результатов из разных веток
|
||||||
|
|
||||||
|
**Подсказка:**
|
||||||
|
- Вместо `print("Обработка CSV...")` загрузите реальный файл (например, `sample_data.csv`) и посчитайте статистику
|
||||||
|
- Результат каждой ветки сохраните в отдельный файл — в `merge_results()` загрузите оба (если существуют) и сравните
|
||||||
|
- Учтите, что при ветвлении выполняется только одна ветка — функция слияния должна обрабатывать случай, когда один из файлов отсутствует
|
||||||
|
|
||||||
**Цель задания:** Применить ветвление в реальном сценарии обработки данных.
|
**Цель задания:** Применить ветвление в реальном сценарии обработки данных.
|
||||||
|
|
||||||
### Задание 3: Динамическое ветвление
|
### Задание 3: Динамическое ветвление
|
||||||
**Сложность:** 🟠 Продвинутая
|
**Сложность:** Продвинутая
|
||||||
**Время выполнения:** 25-30 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Реализуйте ветвление на основе внешних параметров (например, переданных через Variables)
|
- Реализуйте ветвление на основе внешних параметров (например, переданных через Variables)
|
||||||
- Добавьте обработку случая, когда ни одна ветка не подходит
|
- Добавьте обработку случая, когда ни одна ветка не подходит
|
||||||
- Создайте механизм логирования выбранного пути выполнения
|
- Создайте механизм логирования выбранного пути выполнения
|
||||||
|
|
||||||
|
**Подсказка:**
|
||||||
|
- Variables создаются в UI: Admin → Variables. Для чтения в коде: `Variable.get('key', default_var='значение')`
|
||||||
|
- Для случая «ни одна ветка не подходит» добавьте `DummyOperator` как ветку-заглушку
|
||||||
|
- `BranchPythonOperator` должен всегда возвращать существующий `task_id` — иначе DAG упадёт с ошибкой
|
||||||
|
|
||||||
**Цель задания:** Освоить динамическое принятие решений в DAG'ах.
|
**Цель задания:** Освоить динамическое принятие решений в DAG'ах.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## ⚠️ Задания для error_handling_dag.py
|
## Задания для error_handling_dag.py
|
||||||
|
|
||||||
**Цель:** Освоить обработку ошибок и создание отказоустойчивых пайплайнов.
|
**Цель:** Освоить обработку ошибок и создание отказоустойчивых пайплайнов.
|
||||||
|
|
||||||
### Задание 1: Настройка стратегий повторения
|
### Задание 1: Настройка стратегий повторения
|
||||||
**Сложность:** 🟡 Средняя
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 20-25 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Измените параметры `retries` и `retry_delay` для разных задач
|
- Измените параметры `retries` и `retry_delay` для разных задач
|
||||||
- Добавьте экспоненциальную задержку между повторными попытками
|
- Добавьте экспоненциальную задержку между повторными попытками
|
||||||
- Реализуйте кастомный обработчик ошибок для конкретных исключений
|
- Реализуйте кастомный обработчик ошибок для конкретных исключений
|
||||||
|
|
||||||
|
**Подсказка:**
|
||||||
|
- Параметры повторения можно задать на уровне задачи (перекроют `default_args`): `retries`, `retry_delay`, `retry_exponential_backoff`, `max_retry_delay`
|
||||||
|
- Для кастомного обработчика: параметр `on_retry_callback` принимает функцию с аргументом `context`, из которого можно получить номер попытки через `context['ti'].try_number`
|
||||||
|
- Запустите DAG несколько раз и посмотрите в логах, как меняется задержка между попытками
|
||||||
|
|
||||||
**Цель задания:** Научиться настраивать стратегии обработки ошибок.
|
**Цель задания:** Научиться настраивать стратегии обработки ошибок.
|
||||||
|
|
||||||
### Задание 2: Создание комплексной обработки ошибок
|
### Задание 2: Создание комплексной обработки ошибок
|
||||||
**Сложность:** 🟠 Продвинутая
|
**Сложность:** Продвинутая
|
||||||
**Время выполнения:** 30-35 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Добавьте задачи для разных типов ошибок (сетевая ошибка, ошибка данных, системная ошибка)
|
- Добавьте задачи для разных типов ошибок (сетевая ошибка, ошибка данных, системная ошибка)
|
||||||
- Создайте механизм эскалации ошибок (после N неудачных попыток)
|
- Создайте механизм эскалации ошибок (после 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: Мониторинг и логирование
|
### Задание 3: Добавление логирования через модуль logging
|
||||||
**Сложность:** 🟠 Продвинутая
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 25-30 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Добавьте детальное логирование всех этапов выполнения
|
- Замените все `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: Пулы и управление ресурсами
|
### Задание 1: Пулы и управление ресурсами
|
||||||
**Сложность:** 🟡 Средняя
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 20-25 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- В интерфейсе Airflow создайте пул `backup_pool` с 2 слотами
|
- В интерфейсе Airflow создайте пул `backup_pool` с 2 слотами
|
||||||
@@ -317,22 +419,31 @@
|
|||||||
- Назначьте этим задачам параметр `pool="backup_pool"` и настройте `pool_slots` так, чтобы одна из задач занимала 2 слота, а другая — 1
|
- Назначьте этим задачам параметр `pool="backup_pool"` и настройте `pool_slots` так, чтобы одна из задач занимала 2 слота, а другая — 1
|
||||||
- Наблюдайте в UI, что одновременно запускается не более 2 задач из этого пула
|
- Наблюдайте в UI, что одновременно запускается не более 2 задач из этого пула
|
||||||
|
|
||||||
|
**Подсказка:**
|
||||||
|
- Пул создаётся в UI: Admin → Pools → "+"
|
||||||
|
- У `PythonOperator` есть параметры `pool` (имя пула) и `pool_slots` (сколько слотов занимает задача)
|
||||||
|
- Если задача занимает 2 слота из 2 доступных — вторая задача будет ждать, даже если они не связаны зависимостями
|
||||||
|
|
||||||
**Цель задания:** Научиться управлять параллелизмом задач через пулы и `pool_slots`.
|
**Цель задания:** Научиться управлять параллелизмом задач через пулы и `pool_slots`.
|
||||||
|
|
||||||
### Задание 2: Обмен данными через XCom
|
### Задание 2: Обмен данными через XCom
|
||||||
**Сложность:** 🟡 Средняя
|
**Сложность:** Средняя
|
||||||
**Время выполнения:** 20-25 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- В том же DAG добавьте задачу `calculate_metrics` (PythonOperator), которая возвращает словарь с агрегированными показателями, например: `{"total_orders": ..., "avg_amount": ...}`
|
- В том же DAG добавьте задачу `calculate_metrics` (PythonOperator), которая возвращает словарь с агрегированными показателями, например: `{"total_orders": ..., "avg_amount": ...}`
|
||||||
- Добавьте задачу `log_metrics`, которая с помощью `xcom_pull` читает результат `calculate_metrics` и выводит значения в лог
|
- Добавьте задачу `log_metrics`, которая с помощью `xcom_pull` читает результат `calculate_metrics` и выводит значения в лог
|
||||||
- Для одной из задач продемонстрируйте использование XCom в Jinja-шаблоне (например, в `bash_command` или SQL-запросе)
|
- Для одной из задач продемонстрируйте использование 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 и их использование в шаблонах.
|
**Цель задания:** Освоить передачу результатов между задачами через XCom и их использование в шаблонах.
|
||||||
|
|
||||||
### Задание 3: TaskGroup и алертинг
|
### Задание 3: TaskGroup и алертинг
|
||||||
**Сложность:** 🟠 Продвинутая
|
**Сложность:** Продвинутая
|
||||||
**Время выполнения:** 30-35 минут
|
|
||||||
|
|
||||||
**Задача:**
|
**Задача:**
|
||||||
- Объедините логически связанные задачи (например, `extract` / `transform` / `load`) в `TaskGroup`'ы
|
- Объедините логически связанные задачи (например, `extract` / `transform` / `load`) в `TaskGroup`'ы
|
||||||
@@ -340,21 +451,28 @@
|
|||||||
- Настройте для задачи уведомления `trigger_rule=TriggerRule.ALL_DONE`, чтобы уведомление отправлялось даже при частичных ошибках
|
- Настройте для задачи уведомления `trigger_rule=TriggerRule.ALL_DONE`, чтобы уведомление отправлялось даже при частичных ошибках
|
||||||
- При желании вынесите функцию отправки письма в отдельный модуль `utils.py` и импортируйте её в DAG
|
- При желании вынесите функцию отправки письма в отдельный модуль `utils.py` и импортируйте её в DAG
|
||||||
|
|
||||||
|
**Подсказка:**
|
||||||
|
- TaskGroup — контекстный менеджер: `with TaskGroup("имя") as group:` — внутри определяете задачи как обычно
|
||||||
|
- Зависимости работают на уровне групп: `group_a >> group_b`
|
||||||
|
- `TriggerRule.ALL_DONE` (из `airflow.utils.trigger_rule`) означает: «запустись, когда все upstream завершились — неважно, с успехом или ошибкой»
|
||||||
|
- В Graph View группы отображаются как складные блоки — удобно для больших DAG'ов
|
||||||
|
|
||||||
**Цель задания:** Научиться группировать задачи с помощью TaskGroup и строить схему оповещений о статусе пайплайна.
|
**Цель задания:** Научиться группировать задачи с помощью TaskGroup и строить схему оповещений о статусе пайплайна.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## 🎯 Рекомендации по выполнению
|
## Рекомендации по выполнению
|
||||||
|
|
||||||
1. **Начинайте с простых заданий** и постепенно переходите к сложным
|
1. **Начинайте с простых заданий** и постепенно переходите к сложным
|
||||||
2. **Тестируйте каждое изменение** через интерфейс Airflow
|
2. **Тестируйте каждое изменение** через интерфейс Airflow
|
||||||
3. **Изучайте логи выполнения** для понимания поведения задач
|
3. **Изучайте логи выполнения** для понимания поведения задач (вкладка Log у каждой задачи в UI)
|
||||||
4. **Экспериментируйте** с разными настройками и параметрами
|
4. **Экспериментируйте** с разными настройками и параметрами
|
||||||
5. **Документируйте** свои решения и находки
|
5. **Не бойтесь ломать** — DAG'и можно откатить через `git checkout`, а таблицы пересоздать
|
||||||
|
|
||||||
## 📊 Оценка прогресса
|
## Оценка прогресса
|
||||||
|
|
||||||
- 🟢 **Начальный уровень:** Выполнены задания для hello_world_dag и sql_basic_dag
|
- **Начальный уровень:** Выполнены задания для hello_world_dag и sql_basic_dag
|
||||||
- 🟡 **Средний уровень:** Выполнены задания для file_operations_dag, csv_to_postgres.py, csv_to_postgres_dq.py и 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 и алертингу
|
- **Продвинутый уровень:** Выполнены все задания, включая branching_dag, error_handling_dag и дополнительные задания по пулам, XCom, TaskGroup и алертингу
|
||||||
Удачи в изучении Apache Airflow! 🚀
|
|
||||||
|
Удачи в изучении Apache Airflow!
|
||||||
|
|||||||
Reference in New Issue
Block a user