diff --git a/airflow-docker/.gitignore b/airflow-docker/.gitignore new file mode 100644 index 0000000..28db96b --- /dev/null +++ b/airflow-docker/.gitignore @@ -0,0 +1,9 @@ +.vscode/settings.json + +# Do not commit secrets +.env +__pycache__/ +*/__pycache__/ +*.pyc +.venv/ +data/output/ diff --git a/airflow-docker/README.md b/airflow-docker/README.md new file mode 100644 index 0000000..75a75f1 --- /dev/null +++ b/airflow-docker/README.md @@ -0,0 +1,219 @@ +# Educational Airflow Setup for Beginners + +Простой учебный стенд Apache Airflow для начинающих, изучающих SQL и Python. + +## 🎯 Цель проекта + +Создать максимально простую среду для изучения Apache Airflow, где все переменные окружения захардкожены для удобства студентов. + +## 📋 Предварительные требования + +- Docker и Docker Compose +- Базовые знания Python и SQL +- Веб-браузер для доступа к интерфейсу Airflow + +## 🚀 Быстрый старт + +### 1. Клонирование и настройка + +```bash +# Перейдите в директорию проекта +cd airflow-docker + +# Создайте необходимые директории +mkdir -p dags data/input data/output logs +``` + +### 2. Запуск стенда + +```bash +# Запустите все сервисы +docker-compose up -d +``` + +### 3. Доступ к интерфейсам + +- **Airflow UI**: http://localhost:8080 + - Логин: `admin` + - Пароль: `admin` + +- **PostgreSQL для тренировок**: `localhost:5432` + - База данных: `training` + - Пользователь: `student` + - Пароль: `student` + +- **PostgreSQL для метаданных Airflow**: `localhost:5434` + - База данных: `airflow` + - Пользователь: `airflow` + - Пароль: `airflow` + +## 🏗️ Архитектура стенда + +```mermaid +graph TB + A[Airflow Webserver] --> B[PostgreSQL Metadata] + C[Airflow Scheduler] --> B + D[Educational DAGs] --> E[PostgreSQL Training] + D --> F[Local Filesystem] + + A -- порт 8080 --> G[Пользователь] + E -- порт 5432 --> H[SQL Клиент] + + subgraph "Контейнеры Docker" + A + C + B + E + end + + subgraph "Локальная файловая система" + F + D + end +``` + +## 📁 Структура проекта + +``` +airflow-docker/ +├── docker-compose.yml # Конфигурация Docker +├── .env # Переменные окружения (создается автоматически) +├── dags/ # DAG файлы для обучения +│ ├── hello_world_dag.py # Базовый пример +│ ├── sql_basic_dag.py # Работа с SQL +│ ├── file_operations_dag.py # Обработка файлов +│ └── data_processing_dag.py # ETL пайплайн +├── data/ # Данные для упражнений +│ ├── input/ # Входные данные +│ └── output/ # Результаты обработки +├── logs/ # Логи Airflow +└── README.md # Эта инструкция +``` + +## 🎓 Учебные материалы + +### Неделя 1: Основы Airflow + +**Цели:** +- Понимание структуры DAG +- Создание простых задач +- Настройка зависимостей между задачами + +**Примеры DAG:** +- `hello_world_dag.py` - базовые операторы Python +- `sql_basic_dag.py` - работа с базами данных + +### Неделя 2: Интеграция с данными + +**Цели:** +- Подключение к PostgreSQL +- Выполнение SQL запросов +- Обработка файлов CSV + +**Примеры DAG:** +- `file_operations_dag.py` - работа с файлами +- `data_processing_dag.py` - ETL процессы + +### Неделя 3: Продвинутые возможности + +**Цели:** +- Условное выполнение задач +- Обработка ошибок +- Параметризация workflows + +**Примеры DAG:** +- `branching_dag.py` - условная логика +- `error_handling_dag.py` - обработка ошибок + +## 🔧 Технические детали + +### Переменные окружения + +Все переменные захардкожены для простоты: + +```env +# Airflow +AIRFLOW_USER=admin +AIRFLOW_PASSWORD=admin + +# PostgreSQL для метаданных Airflow +POSTGRES_USER=airflow +POSTGRES_PASSWORD=airflow +POSTGRES_DB=airflow + +# PostgreSQL для учебных упражнений +POSTGRES_USER=student +POSTGRES_PASSWORD=student +POSTGRES_DB=training +``` + +### Порты + +- `8080` - Airflow Webserver +- `5432` - PostgreSQL для тренировок +- `5433` - PostgreSQL для метаданных Airflow + +## 🛠️ Управление стендом + +### Запуск сервисов +```bash +docker-compose up -d +``` + +### Остановка сервисов +```bash +docker-compose down +``` + +### Просмотр логов +```bash +# Логи Airflow +docker-compose logs airflow-webserver +docker-compose logs airflow-scheduler + +# Логи PostgreSQL +docker-compose logs postgres-training +docker-compose logs postgres-metadata +``` + +### Перезапуск конкретного сервиса +```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/) +- [Учебные материалы в родительской папке](../) + +## 👥 Для преподавателей + +### Добавление новых DAG +1. Создайте файл Python в папке `dags/` +2. Убедитесь в правильности структуры DAG +3. Файл автоматически появится в интерфейсе Airflow + +### Изменение конфигурации +Все настройки находятся в `docker-compose.yml` и захардкожены для простоты использования. + +### Расширение функциональности +Для добавления новых сервисов или изменения конфигурации отредактируйте `docker-compose.yml`. + +--- + +**Примечание**: Этот стенд предназначен исключительно для учебных целей. Для production использования требуется дополнительная настройка безопасности. diff --git a/airflow-docker/dag-specifications.md b/airflow-docker/dag-specifications.md new file mode 100644 index 0000000..0f39b10 --- /dev/null +++ b/airflow-docker/dag-specifications.md @@ -0,0 +1,245 @@ +# Educational DAG Specifications for Airflow Learning + +## Learning Progression Structure + +### Level 1: Basic Concepts (Week 1) + +#### 1.1 hello_world_dag.py +**Learning Objectives:** +- Understand basic DAG structure +- Learn about PythonOperator +- Understand task dependencies + +**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!") + +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 + +**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 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 + +## 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 + +## 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 \ No newline at end of file diff --git a/airflow-docker/dags/branching_dag.py b/airflow-docker/dags/branching_dag.py new file mode 100644 index 0000000..1f37936 --- /dev/null +++ b/airflow-docker/dags/branching_dag.py @@ -0,0 +1,100 @@ +""" +DAG для демонстрации условного выполнения задач в Airflow +Уровень: Продвинутый +""" +from datetime import datetime, timedelta +from airflow import DAG +from airflow.operators.python import PythonOperator, BranchPythonOperator +from airflow.operators.dummy import DummyOperator +import random + +# Определение DAG +default_args = { + 'owner': 'student', + 'depends_on_past': False, + 'start_date': datetime(2023, 1, 1), + 'email_on_failure': False, + 'email_on_retry': False, + 'retries': 1, + 'retry_delay': timedelta(minutes=5) +} + +dag = DAG( + 'branching_dag', + default_args=default_args, + description='DAG для изучения условного выполнения задач в Airflow', + schedule_interval=timedelta(days=1), + catchup=False, + tags=['educational', 'branching', 'advanced'] +) + +def check_data_quality(): + """Проверка качества данных - случайным образом определяет, какие данные использовать""" + # В реальном сценарии здесь будет проверка качества данных + # Для учебных целей просто случайное решение + quality_score = random.random() # случайное число от 0 до 1 + + if quality_score > 0.5: + print(f"Качество данных хорошее (оценка: {quality_score:.2f}), используем CSV") + return 'process_csv_branch' + else: + print(f"Качество данных требует внимания (оценка: {quality_score:.2f}), используем JSON") + return 'process_json_branch' + +def process_csv_data(): + """Обработка CSV данных""" + print("Обработка CSV файла...") + # Здесь будет логика обработки CSV файла + return "CSV данные обработаны" + +def process_json_data(): + """Обработка JSON данных""" + print("Обработка JSON файла...") + # Здесь будет логика обработки JSON файла + return "JSON данные обработаны" + +def merge_results(): + """Объединение результатов из разных веток""" + print("Объединение результатов из разных веток...") + return "Результаты объединены" + +# Определение задач +start_task = DummyOperator( + task_id='start_task', + dag=dag +) + +check_quality_task = BranchPythonOperator( + task_id='check_data_quality', + python_callable=check_data_quality, + dag=dag +) + +process_csv_task = PythonOperator( + task_id='process_csv_branch', + python_callable=process_csv_data, + dag=dag +) + +process_json_task = PythonOperator( + task_id='process_json_branch', + python_callable=process_json_data, + dag=dag +) + +merge_task = PythonOperator( + task_id='merge_results', + python_callable=merge_results, + trigger_rule='none_failed_or_skipped', # Выполняется, когда одна из веток завершена + dag=dag +) + +end_task = DummyOperator( + task_id='end_task', + dag=dag +) + +# Установка зависимостей +start_task >> check_quality_task +check_quality_task >> [process_csv_task, process_json_task] +[process_csv_task, process_json_task] >> merge_task >> end_task \ No newline at end of file diff --git a/airflow-docker/dags/data_processing_dag.py b/airflow-docker/dags/data_processing_dag.py new file mode 100644 index 0000000..f46b853 --- /dev/null +++ b/airflow-docker/dags/data_processing_dag.py @@ -0,0 +1,186 @@ +""" +DAG для демонстрации ETL процессов в Airflow +Уровень: Средний-Продвинутый +""" +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 = { + 'owner': 'student', + 'depends_on_past': False, + 'start_date': datetime(2023, 1, 1), + 'email_on_failure': False, + 'email_on_retry': False, + 'retries': 1, + 'retry_delay': timedelta(minutes=5) +} + +dag = DAG( + 'data_processing_dag', + default_args=default_args, + description='DAG для изучения ETL процессов в Airflow', + schedule_interval=timedelta(days=1), + catchup=False, + tags=['educational', 'etl', 'intermediate'] +) + +def create_sample_data(): + """Создание примерных данных для ETL процесса""" + # Создаем файлы с данными клиентов и заказов + customers_data = { + 'customer_id': [1, 2, 3, 4, 5], + 'name': ['Alice Johnson', 'Bob Smith', 'Charlie Brown', 'Diana Prince', 'Eve Wilson'], + 'email': ['alice@example.com', 'bob@example.com', 'charlie@example.com', 'diana@example.com', 'eve@example.com'], + 'join_date': ['2023-01-15', '2023-02-20', '2023-03-10', '2023-04-05', '2023-05-12'] + } + + orders_data = { + 'order_id': [101, 102, 103, 104, 105, 106], + 'customer_id': [1, 2, 1, 3, 4, 5], + 'product': ['Laptop', 'Monitor', 'Keyboard', 'Mouse', 'Tablet', 'Headphones'], + 'amount': [1200, 300, 80, 25, 400, 100], + 'order_date': ['2023-10-01', '2023-10-02', '2023-10-03', '2023-10-04', '2023-10-05', '2023-10-06'] + } + + # Сохраняем в CSV + pd.DataFrame(customers_data).to_csv('/opt/airflow/data/input/customers.csv', index=False) + pd.DataFrame(orders_data).to_csv('/opt/airflow/data/input/orders.csv', index=False) + + print("Созданы файлы с данными клиентов и заказов") + return "Sample data created" + +def extract_customers(): + """Извлечение данных клиентов""" + df = pd.read_csv('/opt/airflow/data/input/customers.csv') + print(f"Извлечено {len(df)} записей клиентов") + + # Сохраняем извлеченные данные + df.to_csv('/opt/airflow/data/output/extracted_customers.csv', index=False) + return f"Извлечено {len(df)} клиентов" + +def extract_orders(): + """Извлечение данных заказов""" + df = pd.read_csv('/opt/airflow/data/input/orders.csv') + print(f"Извлечено {len(df)} записей заказов") + + # Сохраняем извлеченные данные + df.to_csv('/opt/airflow/data/output/extracted_orders.csv', index=False) + return f"Извлечено {len(df)} заказов" + +def transform_data(): + """Преобразование данных - объединение клиентов и заказов""" + customers_df = pd.read_csv('/opt/airflow/data/input/customers.csv') + orders_df = pd.read_csv('/opt/airflow/data/input/orders.csv') + + # Объединяем данные + merged_df = pd.merge(orders_df, customers_df, on='customer_id', how='left') + + # Добавляем вычисляемые поля + merged_df['total_spent'] = merged_df['amount'] + merged_df['order_month'] = pd.to_datetime(merged_df['order_date']).dt.month + + # Сохраняем преобразованные данные + merged_df.to_csv('/opt/airflow/data/output/transformed_data.csv', index=False) + print(f"Преобразованы данные: {len(merged_df)} записей") + + return f"Преобразованы {len(merged_df)} записей" + +def load_to_database(): + """Загрузка данных в базу данных (симуляция)""" + df = pd.read_csv('/opt/airflow/data/output/transformed_data.csv') + + # В реальном сценарии здесь был бы код для загрузки в базу данных + # Для учебных целей просто логируем + print(f"Загружено в базу данных: {len(df)} записей") + + # Создаем SQL для создания таблицы (в реальном сценарии) + create_table_sql = """ + CREATE TABLE IF NOT EXISTS customer_orders ( + order_id INTEGER, + customer_id INTEGER, + product VARCHAR(100), + amount DECIMAL(10,2), + order_date DATE, + name VARCHAR(100), + email VARCHAR(100), + join_date DATE, + total_spent DECIMAL(10,2), + order_month INTEGER + ); + """ + + print("SQL для создания таблицы:") + print(create_table_sql) + + return f"Подготовлено к загрузке в базу: {len(df)} записей" + +def generate_report(): + """Генерация отчета""" + df = pd.read_csv('/opt/airflow/data/output/transformed_data.csv') + + # Создаем простой отчет + report = { + 'total_orders': len(df), + 'total_revenue': df['amount'].sum(), + 'avg_order_value': df['amount'].mean(), + 'unique_customers': df['customer_id'].nunique(), + 'top_customer': df.groupby('name')['amount'].sum().idxmax(), + 'top_customer_spending': df.groupby('name')['amount'].sum().max() + } + + # Сохраняем отчет в файл + with open('/opt/airflow/data/output/report.txt', 'w') as f: + f.write("Отчет по заказам клиентов\n") + f.write("=" * 30 + "\n") + f.write(f"Всего заказов: {report['total_orders']}\n") + f.write(f"Общая выручка: ${report['total_revenue']}\n") + f.write(f"Средний чек: ${report['avg_order_value']:.2f}\n") + f.write(f"Уникальных клиентов: {report['unique_customers']}\n") + f.write(f"Лучший клиент: {report['top_customer']} (${report['top_customer_spending']})\n") + + print("Создан отчет по заказам") + return "Отчет создан" + +# Определение задач +create_data_task = PythonOperator( + task_id='create_sample_data', + python_callable=create_sample_data, + dag=dag +) + +extract_customers_task = PythonOperator( + task_id='extract_customers', + python_callable=extract_customers, + dag=dag +) + +extract_orders_task = PythonOperator( + task_id='extract_orders', + python_callable=extract_orders, + dag=dag +) + +transform_task = PythonOperator( + task_id='transform_data', + python_callable=transform_data, + dag=dag +) + +load_task = PythonOperator( + task_id='load_to_database', + python_callable=load_to_database, + dag=dag +) + +report_task = PythonOperator( + task_id='generate_report', + python_callable=generate_report, + dag=dag +) + +# Установка зависимостей +create_data_task >> [extract_customers_task, extract_orders_task] >> transform_task >> load_task >> report_task \ No newline at end of file diff --git a/airflow-docker/dags/error_handling_dag.py b/airflow-docker/dags/error_handling_dag.py new file mode 100644 index 0000000..abaabcc --- /dev/null +++ b/airflow-docker/dags/error_handling_dag.py @@ -0,0 +1,107 @@ +""" +DAG для демонстрации обработки ошибок в Airflow +Уровень: Продвинутый +""" +from datetime import datetime, timedelta +from airflow import DAG +from airflow.operators.python import PythonOperator +from airflow.operators.dummy import DummyOperator +import random + +# Определение DAG +default_args = { + 'owner': 'student', + 'depends_on_past': False, + 'start_date': datetime(2023, 1, 1), + 'email_on_failure': False, # Отключаем email уведомления для простоты + 'email_on_retry': False, + 'retries': 3, # Количество попыток при ошибке + 'retry_delay': timedelta(seconds=10) # Задержка между попытками +} + +dag = DAG( + 'error_handling_dag', + default_args=default_args, + description='DAG для изучения обработки ошибок в Airflow', + schedule_interval=timedelta(days=1), + catchup=False, + tags=['educational', 'error_handling', 'advanced'] +) + +def unreliable_task(): + """Задача, которая может завершиться с ошибкой""" + # В реальном сценарии это может быть задача, зависящая от внешних факторов + # Для учебных целей случайным образом генерируем ошибку + if random.random() < 0.3: # 30% вероятность ошибки + print("Ошибка: задача не выполнена успешно!") + raise Exception("Случайная ошибка в задаче") + + print("Задача выполнена успешно!") + return "Задача выполнена" + +def success_handler(): + """Обработчик успешного выполнения""" + print("Поздравляем! Все задачи выполнены успешно!") + return "Успешно завершено" + +def failure_handler(): + """Обработчик ошибок""" + print("Одна или несколько задач завершились с ошибкой!") + print("Проверьте логи для получения дополнительной информации") + return "Ошибка обработана" + +def retry_task(): + """Задача с механизмом повторных попыток""" + # Имитируем задачу, которая может завершиться с ошибкой, но со временем исправляется + import time + time.sleep(2) # Имитация работы + + # С вероятностью 50% задача завершится с ошибкой + if random.random() < 0.5: + print("Ошибка в retry_task!") + raise Exception("Ошибка в задаче с повторными попытками") + + print("retry_task выполнена успешно!") + return "retry_task завершена" + +# Определение задач +start_task = DummyOperator( + task_id='start_task', + dag=dag +) + +unreliable_task = PythonOperator( + task_id='unreliable_task', + python_callable=unreliable_task, + dag=dag +) + +retry_task = PythonOperator( + task_id='retry_task', + python_callable=retry_task, + dag=dag +) + +success_handler_task = PythonOperator( + task_id='success_handler', + python_callable=success_handler, + trigger_rule='all_success', # Выполняется только если все предыдущие задачи успешны + dag=dag +) + +failure_handler_task = PythonOperator( + task_id='failure_handler', + python_callable=failure_handler, + trigger_rule='one_failed', # Выполняется если хотя бы одна предыдущая задача завершилась с ошибкой + dag=dag +) + +end_task = DummyOperator( + task_id='end_task', + dag=dag +) + +# Установка зависимостей +start_task >> [unreliable_task, retry_task] +[unreliable_task, retry_task] >> [success_handler_task, failure_handler_task] +[success_handler_task, failure_handler_task] >> end_task \ No newline at end of file diff --git a/airflow-docker/dags/file_operations_dag.py b/airflow-docker/dags/file_operations_dag.py new file mode 100644 index 0000000..7385242 --- /dev/null +++ b/airflow-docker/dags/file_operations_dag.py @@ -0,0 +1,125 @@ +""" +DAG для демонстрации работы с файлами в Airflow +Уровень: Средний +""" +import pandas as pd +import os +from datetime import datetime, timedelta +from airflow import DAG +from airflow.operators.python import PythonOperator +from airflow.operators.bash import BashOperator + +# Определение DAG +default_args = { + 'owner': 'student', + 'depends_on_past': False, + 'start_date': datetime(2023, 1, 1), + 'email_on_failure': False, + 'email_on_retry': False, + 'retries': 1, + 'retry_delay': timedelta(minutes=5) +} + +dag = DAG( + 'file_operations_dag', + default_args=default_args, + description='DAG для изучения работы с файлами в Airflow', + schedule_interval=timedelta(days=1), + catchup=False, + tags=['educational', 'files', 'intermediate'] +) + +def generate_sample_data(): + """Создание CSV файла с примерными данными""" + import pandas as pd + import random + + data = { + 'id': range(1, 101), + 'name': [f'User_{i}' for i in range(1, 101)], + 'age': [random.randint(18, 65) for _ in range(100)], + 'salary': [random.randint(30000, 100000) for _ in range(100)] + } + + df = pd.DataFrame(data) + df.to_csv('/opt/airflow/data/input/sample_data.csv', index=False) + print(f"Создан файл с {len(df)} записями") + return "sample_data.csv created" + +def read_and_validate_data(): + """Чтение и валидация CSV файла""" + df = pd.read_csv('/opt/airflow/data/input/sample_data.csv') + print(f"Прочитан файл: {len(df)} строк, {len(df.columns)} столбцов") + + # Простая валидация + assert len(df) > 0, "Файл пустой" + assert 'name' in df.columns, "Отсутствует столбец name" + + return f"Файл валидирован: {len(df)} записей" + +def transform_data(): + """Преобразование данных""" + df = pd.read_csv('/opt/airflow/data/input/sample_data.csv') + + # Простое преобразование - добавим столбец с категорией зарплаты + df['salary_category'] = df['salary'].apply( + lambda x: 'High' if x >= 70000 else 'Medium' if x >= 50000 else 'Low' + ) + + # Сохраняем обработанные данные + df.to_csv('/opt/airflow/data/output/processed_data.csv', index=False) + print(f"Обработаны данные: {len(df)} записей") + + return f"Данные обработаны: {len(df)} записей" + +def write_summary(): + """Создание сводки по обработанным данным""" + df = pd.read_csv('/opt/airflow/data/output/processed_data.csv') + + summary = { + 'total_records': len(df), + 'avg_salary': df['salary'].mean(), + 'high_salary_count': len(df[df['salary_category'] == 'High']), + 'medium_salary_count': len(df[df['salary_category'] == 'Medium']), + 'low_salary_count': len(df[df['salary_category'] == 'Low']) + } + + # Сохраняем сводку в текстовый файл + with open('/opt/airflow/data/output/summary.txt', 'w') as f: + f.write("Сводка по обработанным данным:\n") + f.write(f"Всего записей: {summary['total_records']}\n") + f.write(f"Средняя зарплата: {summary['avg_salary']:.2f}\n") + f.write(f"Высокая зарплата: {summary['high_salary_count']}\n") + f.write(f"Средняя зарплата: {summary['medium_salary_count']}\n") + f.write(f"Низкая зарплата: {summary['low_salary_count']}\n") + + print("Создана сводка по данным") + return "Сводка создана" + +# Определение задач +generate_task = PythonOperator( + task_id='generate_sample_data', + python_callable=generate_sample_data, + dag=dag +) + +read_task = PythonOperator( + task_id='read_and_validate_data', + python_callable=read_and_validate_data, + dag=dag +) + +transform_task = PythonOperator( + task_id='transform_data', + python_callable=transform_data, + dag=dag +) + +summary_task = PythonOperator( + task_id='write_summary', + python_callable=write_summary, + dag=dag +) + +# Установка зависимостей +generate_task >> read_task >> transform_task >> summary_task \ No newline at end of file diff --git a/airflow-docker/dags/hello_world_dag.py b/airflow-docker/dags/hello_world_dag.py new file mode 100644 index 0000000..f1cf6b0 --- /dev/null +++ b/airflow-docker/dags/hello_world_dag.py @@ -0,0 +1,63 @@ +""" +Простой DAG для демонстрации основных концепций Airflow +Уровень: Начальный +""" +from datetime import datetime, timedelta +from airflow import DAG +from airflow.operators.python import PythonOperator +from airflow.operators.bash import BashOperator + +# Определение DAG +default_args = { + 'owner': 'student', + 'depends_on_past': False, + 'start_date': datetime(2023, 1, 1), + 'email_on_failure': False, + 'email_on_retry': False, + 'retries': 1, + 'retry_delay': timedelta(minutes=5) +} + +dag = DAG( + 'hello_world_dag', + default_args=default_args, + description='Простой DAG для изучения основ Airflow', + schedule_interval=timedelta(days=1), + catchup=False, + tags=['educational', 'beginner'] +) + +# Функции для задач +def print_hello(): + print("Hello World from Airflow!") + return 'Hello World!' + +def print_date(): + print(f"Current date: {datetime.now()}") + return f"Date: {datetime.now()}" + +def print_goodbye(): + print("Goodbye from Airflow!") + return 'Goodbye!' + +# Определение задач +start_task = PythonOperator( + task_id='start_task', + python_callable=print_hello, + dag=dag +) + +date_task = PythonOperator( + task_id='date_task', + python_callable=print_date, + dag=dag +) + +end_task = PythonOperator( + task_id='end_task', + python_callable=print_goodbye, + dag=dag +) + +# Установка зависимостей +start_task >> date_task >> end_task \ No newline at end of file diff --git a/airflow-docker/dags/sql_basic_dag.py b/airflow-docker/dags/sql_basic_dag.py new file mode 100644 index 0000000..6050fe2 --- /dev/null +++ b/airflow-docker/dags/sql_basic_dag.py @@ -0,0 +1,86 @@ +""" +DAG для демонстрации работы с SQL в Airflow +Уровень: Начальный-Средний +""" +from datetime import datetime, timedelta +from airflow import DAG +from airflow.operators.python import PythonOperator +from airflow.providers.postgres.operators.postgres import PostgresOperator + +# Определение DAG +default_args = { + 'owner': 'student', + 'depends_on_past': False, + 'start_date': datetime(2023, 1, 1), + 'email_on_failure': False, + 'email_on_retry': False, + 'retries': 1, + 'retry_delay': timedelta(minutes=5) +} + +dag = DAG( + 'sql_basic_dag', + default_args=default_args, + description='DAG для изучения SQL операций в Airflow', + schedule_interval=timedelta(days=1), + catchup=False, + tags=['educational', 'sql', 'beginner'] +) + +# SQL команды +create_table_sql = """ +CREATE TABLE IF NOT EXISTS students ( + id SERIAL PRIMARY KEY, + name VARCHAR(100), + age INTEGER, + created_at TIMESTAMP DEFAULT NOW() +); +""" + +insert_data_sql = """ +INSERT INTO students (name, age) VALUES +('Alice', 22), +('Bob', 24), +('Charlie', 21) +ON CONFLICT DO NOTHING; +""" + +query_data_sql = """ +SELECT * FROM students; +""" + +drop_table_sql = """ +-- DROP TABLE IF EXISTS students; -- Закомментировано для сохранения данных +""" + +# Определение задач +create_table_task = PostgresOperator( + task_id='create_table', + postgres_conn_id='postgres_training', # Это соединение нужно будет создать вручную в Airflow UI + sql=create_table_sql, + dag=dag +) + +insert_data_task = PostgresOperator( + task_id='insert_data', + postgres_conn_id='postgres_training', + sql=insert_data_sql, + dag=dag +) + +query_data_task = PostgresOperator( + task_id='query_data', + postgres_conn_id='postgres_training', + sql=query_data_sql, + dag=dag +) + +drop_table_task = PostgresOperator( + task_id='drop_table', + postgres_conn_id='postgres_training', + sql=drop_table_sql, + dag=dag +) + +# Установка зависимостей +create_table_task >> insert_data_task >> query_data_task >> drop_table_task \ No newline at end of file diff --git a/airflow-docker/data/input/customers.csv b/airflow-docker/data/input/customers.csv new file mode 100644 index 0000000..ad64cc7 --- /dev/null +++ b/airflow-docker/data/input/customers.csv @@ -0,0 +1,6 @@ +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 +4,Diana Prince,diana@example.com,2023-04-05 +5,Eve Wilson,eve@example.com,2023-05-12 \ No newline at end of file diff --git a/airflow-docker/data/input/orders.csv b/airflow-docker/data/input/orders.csv new file mode 100644 index 0000000..e3b9d53 --- /dev/null +++ b/airflow-docker/data/input/orders.csv @@ -0,0 +1,7 @@ +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 +105,4,Tablet,400,2023-10-05 +106,5,Headphones,100,2023-10-06 \ No newline at end of file diff --git a/airflow-docker/docker-compose.yml b/airflow-docker/docker-compose.yml new file mode 100644 index 0000000..d74328b --- /dev/null +++ b/airflow-docker/docker-compose.yml @@ -0,0 +1,104 @@ +version: '3.8' + +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 + healthcheck: + test: ["CMD-SHELL", "pg_isready -U airflow -d airflow"] + interval: 5s + timeout: 5s + retries: 20 + + # 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 + - ./init-postgres.sql:/docker-entrypoint-initdb.d/init-postgres.sql + healthcheck: + test: ["CMD-SHELL", "pg_isready -U student -d training"] + interval: 5s + timeout: 5s + retries: 20 + + # Airflow webserver + 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 + AIRFLOW__CORE__EXECUTOR: LocalExecutor + command: > + bash -c " + pip install --no-cache-dir psycopg2-binary && + airflow webserver + " + ports: + - "8080:8080" + volumes: + - ./dags:/opt/airflow/dags + - ./data:/opt/airflow/data + depends_on: + postgres-metadata: + condition: service_healthy + postgres-training: + condition: service_healthy + + # Airflow scheduler + 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 + AIRFLOW__CORE__EXECUTOR: LocalExecutor + command: > + bash -c " + pip install --no-cache-dir psycopg2-binary && + airflow scheduler + " + volumes: + - ./dags:/opt/airflow/dags + - ./data:/opt/airflow/data + depends_on: + postgres-metadata: + condition: service_healthy + postgres-training: + condition: service_healthy + + # Airflow init to create admin user + 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 + volumes: + - ./dags:/opt/airflow/dags + - ./data:/opt/airflow/data + command: > + bash -c " + pip install --no-cache-dir psycopg2-binary && + airflow db migrate && + airflow users create --username admin --password admin --firstname Admin --lastname User --role Admin --email admin@example.org + " + depends_on: + postgres-metadata: + condition: service_healthy + +volumes: + pgmeta: + pg_data: diff --git a/airflow-docker/educational-setup-plan.md b/airflow-docker/educational-setup-plan.md new file mode 100644 index 0000000..29ec479 --- /dev/null +++ b/airflow-docker/educational-setup-plan.md @@ -0,0 +1,150 @@ +# 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) +- Missing Greenplum service (referenced in dependencies but not defined) +- 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: + - "5433: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** +- `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 \ No newline at end of file diff --git a/airflow-docker/init-postgres.sql b/airflow-docker/init-postgres.sql new file mode 100644 index 0000000..03ecaf8 --- /dev/null +++ b/airflow-docker/init-postgres.sql @@ -0,0 +1,59 @@ +-- Инициализационный скрипт для PostgreSQL для учебных упражнений + +-- Создание таблицы студентов +CREATE TABLE IF NOT EXISTS students ( + student_id SERIAL PRIMARY KEY, + first_name VARCHAR(50), + last_name VARCHAR(50), + email VARCHAR(100), + enrollment_date DATE, + grade INTEGER +); + +-- Создание таблицы курсов +CREATE TABLE IF NOT EXISTS courses ( + course_id SERIAL PRIMARY KEY, + course_name VARCHAR(100), + instructor VARCHAR(100), + credits INTEGER +); + +-- Создание таблицы зачислений +CREATE TABLE IF NOT EXISTS 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) +); + +-- Вставка тестовых данных +INSERT INTO students (first_name, last_name, email, enrollment_date, grade) VALUES +('Alice', 'Johnson', 'alice@example.com', '2023-09-01', 85), +('Bob', 'Smith', 'bob@example.com', '2023-09-01', 92), +('Charlie', 'Brown', 'charlie@example.com', '2023-09-01', 78), +('Diana', 'Prince', 'diana@example.com', '2023-09-01', 96), +('Eve', 'Wilson', 'eve@example.com', '2023-09-01', 8) +ON CONFLICT DO NOTHING; + +INSERT INTO courses (course_name, instructor, credits) VALUES +('Introduction to Data Science', 'Dr. Smith', 3), +('Python Programming', 'Dr. Johnson', 4), +('Database Systems', 'Dr. Williams', 3), +('Machine Learning', 'Dr. Brown', 4) +ON CONFLICT DO NOTHING; + +INSERT INTO enrollments (student_id, course_id, enrollment_date, grade) VALUES +(1, 1, '2023-09-01', 'A'), +(1, 2, '2023-09-01', 'B'), +(2, 1, '2023-09-01', 'A'), +(2, 3, '2023-09-01', 'A'), +(3, 2, '2023-09-01', 'B'), +(4, 4, '2023-09-01', 'A'), +(5, 1, '2023-09-01', 'A') +ON CONFLICT DO NOTHING; + +-- Создание индексов для улучшения производительности +CREATE INDEX IF NOT EXISTS idx_students_email ON students(email); +CREATE INDEX IF NOT EXISTS idx_enrollments_student ON enrollments(student_id); +CREATE INDEX IF NOT EXISTS idx_enrollments_course ON enrollments(course_id); \ No newline at end of file diff --git a/airflow-docker/requirements.txt b/airflow-docker/requirements.txt new file mode 100644 index 0000000..1a31267 --- /dev/null +++ b/airflow-docker/requirements.txt @@ -0,0 +1,13 @@ +# Core packages for Airflow +psycopg2-binary==2.9.7 +pandas==2.1.4 +numpy==1.24.3 + +# Additional packages for data processing +openpyxl==3.1.2 +xlrd==2.0.1 +python-dateutil==2.8.2 + +# For additional operators (if needed) +apache-airflow-providers-postgres==8.5.1 +apache-airflow-providers-common-sql==1.13.0 \ No newline at end of file