refactor(dags): перенесён pandas в функции, исправлен баг occupations, добавлены docstring

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

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
ddadminandClaude Opus 4.6 committed 2026-03-26 22:50:55 +03:00
1 parent eaea33e761
commit 72d5968d01
5 files changed
+54 -159

No files matched your search

+15 -1
View File
@@ -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 from __future__ import annotations
import logging import logging
@@ -5,7 +17,7 @@ import os
import random import random
from datetime import UTC, datetime, timedelta from datetime import UTC, datetime, timedelta
from pathlib import Path from pathlib import Path
import pandas as pd
from airflow.operators.python import PythonOperator from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook 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: def _generate_csv(rows: int, csv_dir: Path) -> str:
"""Генерирует CSV c заказами с помощью pandas и сохраняет на диск.""" """Генерирует CSV c заказами с помощью pandas и сохраняет на диск."""
import pandas as pd
csv_dir.mkdir(parents=True, exist_ok=True) csv_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S") timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
csv_path = csv_dir / f"orders_{timestamp}.csv" 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: def _preview_csv(csv_path: str, sample_rows: int = 5) -> None:
"""Отображает предпросмотр CSV через pandas (head и describe).""" """Отображает предпросмотр CSV через pandas (head и describe)."""
import pandas as pd
df = pd.read_csv(csv_path) df = pd.read_csv(csv_path)
df["order_ts"] = pd.to_datetime(df["order_ts"], errors="coerce") df["order_ts"] = pd.to_datetime(df["order_ts"], errors="coerce")
logging.info( logging.info(
+13
View File
@@ -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 from __future__ import annotations
import logging import logging
+15 -1
View File
@@ -1,12 +1,20 @@
""" """
DAG для демонстрации ETL процессов в Airflow 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 datetime import datetime, timedelta
from airflow import DAG from airflow import DAG
from airflow.operators.python import PythonOperator from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator from airflow.providers.postgres.operators.postgres import PostgresOperator
import pandas as pd
# Определение DAG # Определение DAG
default_args = { default_args = {
@@ -30,6 +38,7 @@ dag = DAG(
def create_sample_data(): def create_sample_data():
"""Создание примерных данных для ETL процесса""" """Создание примерных данных для ETL процесса"""
import pandas as pd
# Создаем файлы с данными клиентов и заказов # Создаем файлы с данными клиентов и заказов
customers_data = { customers_data = {
'customer_id': [1, 2, 3, 4, 5], 'customer_id': [1, 2, 3, 4, 5],
@@ -55,6 +64,7 @@ def create_sample_data():
def extract_customers(): def extract_customers():
"""Извлечение данных клиентов""" """Извлечение данных клиентов"""
import pandas as pd
df = pd.read_csv('/opt/airflow/data/input/customers.csv') df = pd.read_csv('/opt/airflow/data/input/customers.csv')
print(f"Извлечено {len(df)} записей клиентов") print(f"Извлечено {len(df)} записей клиентов")
@@ -64,6 +74,7 @@ def extract_customers():
def extract_orders(): def extract_orders():
"""Извлечение данных заказов""" """Извлечение данных заказов"""
import pandas as pd
df = pd.read_csv('/opt/airflow/data/input/orders.csv') df = pd.read_csv('/opt/airflow/data/input/orders.csv')
print(f"Извлечено {len(df)} записей заказов") print(f"Извлечено {len(df)} записей заказов")
@@ -73,6 +84,7 @@ def extract_orders():
def transform_data(): def transform_data():
"""Преобразование данных - объединение клиентов и заказов""" """Преобразование данных - объединение клиентов и заказов"""
import pandas as pd
customers_df = pd.read_csv('/opt/airflow/data/input/customers.csv') customers_df = pd.read_csv('/opt/airflow/data/input/customers.csv')
orders_df = pd.read_csv('/opt/airflow/data/input/orders.csv') orders_df = pd.read_csv('/opt/airflow/data/input/orders.csv')
@@ -91,6 +103,7 @@ def transform_data():
def load_to_database(): def load_to_database():
"""Загрузка данных в базу данных (симуляция)""" """Загрузка данных в базу данных (симуляция)"""
import pandas as pd
df = pd.read_csv('/opt/airflow/data/output/transformed_data.csv') df = pd.read_csv('/opt/airflow/data/output/transformed_data.csv')
# В реальном сценарии здесь был бы код для загрузки в базу данных # В реальном сценарии здесь был бы код для загрузки в базу данных
@@ -120,6 +133,7 @@ def load_to_database():
def generate_report(): def generate_report():
"""Генерация отчета""" """Генерация отчета"""
import pandas as pd
df = pd.read_csv('/opt/airflow/data/output/transformed_data.csv') df = pd.read_csv('/opt/airflow/data/output/transformed_data.csv')
# Создаем простой отчет # Создаем простой отчет
+11 -6
View File
@@ -2,13 +2,14 @@
DAG для демонстрации работы с файлами в Airflow 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 import os
from datetime import datetime, timedelta from datetime import datetime, timedelta
from airflow import DAG from airflow import DAG
@@ -56,6 +57,7 @@ def generate_sample_data():
'id': range(1, 101), 'id': range(1, 101),
'name': names, 'name': names,
'age': ages, 'age': ages,
'occupation': occupations,
'salary': [max(30000, min(150000, ages[i] * 1200 + (hash(names[i]) % 15000) - 5000)) for i in range(100)], '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)] 'address': [address.address() for _ in range(100)]
} }
@@ -68,6 +70,7 @@ def generate_sample_data():
def read_and_validate_data(): def read_and_validate_data():
"""Чтение и валидация CSV файла""" """Чтение и валидация CSV файла"""
import pandas as pd
df = pd.read_csv('/opt/airflow/data/input/sample_data.csv') df = pd.read_csv('/opt/airflow/data/input/sample_data.csv')
print(f"Прочитан файл: {len(df)} строк, {len(df.columns)} столбцов") print(f"Прочитан файл: {len(df)} строк, {len(df.columns)} столбцов")
print(f"Колонки: {list(df.columns)}") print(f"Колонки: {list(df.columns)}")
@@ -87,6 +90,7 @@ def read_and_validate_data():
def transform_data(): def transform_data():
"""Преобразование данных""" """Преобразование данных"""
import pandas as pd
import os import os
df = pd.read_csv('/opt/airflow/data/input/sample_data.csv') df = pd.read_csv('/opt/airflow/data/input/sample_data.csv')
@@ -115,6 +119,7 @@ def transform_data():
def write_summary(): def write_summary():
"""Создание сводки по обработанным данным""" """Создание сводки по обработанным данным"""
import pandas as pd
df = pd.read_csv('/opt/airflow/data/output/processed_data.csv') df = pd.read_csv('/opt/airflow/data/output/processed_data.csv')
summary = { summary = {
-151
View File
@@ -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