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