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:
1 parent
eaea33e761
commit
72d5968d01
5 files changed
+54
-159
No files matched your search
@@ -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(
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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')
|
||||||
|
|
||||||
# Создаем простой отчет
|
# Создаем простой отчет
|
||||||
|
|||||||
@@ -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 = {
|
||||||
|
|||||||
@@ -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
|
|
||||||
Reference in new issue
Block a user