Final Project: Retail ETL Pipeline (Eldar)

Dockerfile

@@ -0,0 +1,11 @@
+# Используем официальный образ
+FROM apache/airflow:2.9.2
+
+# Копируем наш файл зависимостей
+COPY pyproject.toml .
+
+# ВАЖНО: Мы НЕ переключаемся на root. Мы остаемся пользователем airflow.
+# 1. Устанавливаем uv через обычный pip
+# 2. Устанавливаем наши библиотеки через uv
+RUN pip install --no-cache-dir uv && \
+    uv pip install --no-cache -r pyproject.toml

dags/Project.py

@@ -0,0 +1,150 @@
+import logging
+from datetime import datetime, timedelta
+import os
+import pandas as pd
+
+from airflow import DAG
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
+from airflow.providers.postgres.operators.postgres import PostgresOperator
+from airflow.providers.postgres.hooks.postgres import PostgresHook
+
+# Настраиваем логгер
+logger = logging.getLogger("airflow.task")
+
+POSTGRES_CONN_ID = 'postgres_etl_target_conn'
+
+def notify_failure(context):
+    """
+    Отправляет уведомление при сбое задачи (Имитация).
+    """
+    task_instance = context.get('task_instance')
+    task_id = task_instance.task_id
+    log_url = task_instance.log_url
+
+    # Имитация отправки алерта
+    logger.error(f"""
+    ###################################################
+    ALARM! ALARM! PIPELINE FAILED!
+    Task: {task_id}
+    Date: {context.get('ds')}
+    Log URL: {log_url}
+    Sending notification to Data Engineering Team...
+    ###################################################
+    """)
+
+def load_csv_to_postgres(execution_date, **kwargs):
+    """
+    Загружает ежедневный CSV в таблицу Staging с маппингом колонок.
+    """
+    date_str = execution_date
+    file_path = f"/opt/airflow/data/sales_{date_str}.csv"
+
+    logger.info(f"Starting extraction for date: {date_str}")
+
+
+    try:
+        if not os.path.exists(file_path):
+            logger.warning(f"File {file_path} not found. Skipping load.")
+            return
+
+        df = pd.read_csv(file_path)
+        logger.info(f"Read {len(df)} rows from CSV.")
+
+        #Маппинг колонок CSV в схему БД (CamelCase -> snake_case)
+        rename_map = {
+            'Invoice': 'invoice_no',
+            'StockCode': 'stock_code',
+            'Description': 'description',
+            'Quantity': 'quantity',
+            'InvoiceDate': 'invoice_date',
+            'Price': 'unit_price',
+            'Customer ID': 'customer_id',
+            'Country': 'country'
+        }
+        df.rename(columns=rename_map, inplace=True)
+        df.columns = [c.lower() for c in df.columns]
+
+        hook = PostgresHook(postgres_conn_id=POSTGRES_CONN_ID)
+        engine = hook.get_sqlalchemy_engine()
+
+        #Замена данных в staging для текущего батча (if_exists='replace')
+        df.to_sql('stage_sales', engine, schema='star', if_exists='replace', index=False)
+        logger.info("Successfully loaded data to stage_sales.")
+
+    except Exception as e:
+        logger.error(f"Critical error in load_csv_to_postgres: {e}")
+        raise e
+
+# Определение DAG
+default_args = {
+    'owner': 'airflow',
+    'depends_on_past': False,
+    'retries': 1,
+    'retry_delay': timedelta(minutes=1),
+    'on_failure_callback': notify_failure, 
+}
+
+with DAG(
+    dag_id='retail_etl_pipeline_v2', # Версия 2 (Улучшенная)
+    default_args=default_args,
+    description='ETL with Star Schema, Logging and Alerting',
+    schedule_interval='@daily', 
+    start_date=datetime(2010, 12, 1),
+    catchup=False,
+    tags=['retail', 'star_schema', 'final_project'],
+) as dag:
+
+    # 1. Извлечение данных (Mock API)
+    extract_data = BashOperator(
+        task_id='extract_data',
+        bash_command='python /opt/airflow/dags/scripts/generate_data.py --date {{ ds }}'
+    )
+
+    # 2. Загрузка сырых данных в Staging
+    load_staging = PythonOperator(
+        task_id='load_staging',
+        python_callable=load_csv_to_postgres,
+        op_kwargs={'execution_date': '{{ ds }}'}
+    )
+
+    # 3. Наполнение Измерений (Логика Upsert / Игнорирование дублей)
+    load_dims = PostgresOperator(
+        task_id='load_dims',
+        postgres_conn_id=POSTGRES_CONN_ID,
+        sql="""
+            INSERT INTO star.dim_products (stock_code, description)
+            SELECT DISTINCT stock_code, description FROM star.stage_sales
+            ON CONFLICT (stock_code) DO NOTHING;
+
+            INSERT INTO star.dim_customers (customer_id, country)
+            SELECT DISTINCT CAST(customer_id AS VARCHAR), country FROM star.stage_sales
+            WHERE customer_id IS NOT NULL
+            ON CONFLICT (customer_id) DO NOTHING;
+        """
+    )
+
+    # 4. Наполнение Фактов (Идемпотентная загрузка)
+    load_facts = PostgresOperator(
+        task_id='load_facts',
+        postgres_conn_id=POSTGRES_CONN_ID,
+        sql="""
+            DELETE FROM star.fact_sales 
+            WHERE invoice_date::DATE = '{{ ds }}'::DATE;
+
+            INSERT INTO star.fact_sales (invoice_no, invoice_date, customer_id, stock_code, quantity, unit_price, total_amount)
+            SELECT 
+                invoice_no,
+                CAST(invoice_date AS TIMESTAMP),
+                CAST(customer_id AS VARCHAR),
+                stock_code,
+                quantity,
+                unit_price,
+                quantity * unit_price
+            FROM star.stage_sales
+            WHERE customer_id IS NOT NULL
+            ON CONFLICT (invoice_no, stock_code, customer_id) DO NOTHING;
+        """
+    )
+
+    extract_data >> load_staging >> load_dims >> load_facts

dags/scripts/generate_data.py

@@ -0,0 +1,59 @@
+import pandas as pd
+import os
+import argparse
+import logging
+
+# Настройка логирования для скрипта
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+def extract_daily_data(target_date):
+    """
+    Читает большой исторический CSV и извлекает данные за указанную дату (target_date).
+    Эмулирует ответ API для ежедневной пакетной загрузки.
+    """
+    base_path = "/opt/airflow/data"
+    source_file = os.path.join(base_path, "retail.csv")
+    output_file = os.path.join(base_path, f"sales_{target_date}.csv")
+
+    logging.info(f"Начало извлечения данных за дату: {target_date}")    
+
+
+    if not os.path.exists(source_file):
+        logging.error(f"Исходный файл {source_file} не найден.")
+        return
+
+    try:
+        # Используем кодировку ISO-8859-1 для корректной обработки спецсимволов
+        df = pd.read_csv(source_file, encoding='ISO-8859-1')
+
+        # 3. Определяем колонку с датой
+        date_col = 'InvoiceDate'
+        if date_col not in df.columns:
+            # Пробуем альтернативы, если файл другой
+            possible_cols = ['Order Date', 'Date', 'invoice_date']
+            for col in possible_cols:
+                if col in df.columns:
+                    date_col = col
+                    break
+
+
+        # Преобразование в datetime и фильтрация по целевой дате
+        df[date_col] = pd.to_datetime(df[date_col])
+        target_dt = pd.to_datetime(target_date).date()
+        daily_data = df[df[date_col].dt.date == target_dt]
+
+        if daily_data.empty:
+            logging.warning(f"Транзакции за {target_date} не найдены")
+        else:
+            daily_data.to_csv(output_file, index=False)
+            logging.info(f"Успешно извлечено {len(daily_data)} строк в файл {output_file}")
+
+    except Exception as e:
+        logging.error(f"Ошибка обработки данных: {e}")
+        raise e
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--date", required=True, help="YYYY-MM-DD")
+    args = parser.parse_args()
+    extract_daily_data(args.date)

dags/scripts/init_db.py

@@ -0,0 +1,86 @@
+import psycopg2
+import os
+
+def create_tables():
+    """
+    Инициализирует структуру схемы 'Звезда' (Star Schema) в PostgreSQL.
+    Создает таблицу Staging, таблицы Измерений (Dimensions) и Фактов (Facts)
+    """
+    conn_params = {
+        "host": "postgres-etl-target",
+        "database": "etl_db",
+        "user": "etl_user",
+        "password": "etl_pass",
+        "port": "5432"
+    }
+
+
+    commands = [
+        # 1. Создание схемы
+        "CREATE SCHEMA IF NOT EXISTS star;",
+
+        # 2. Таблица STAGING (Сырые данные, копия CSV)
+        """
+        CREATE TABLE IF NOT EXISTS star.stage_sales (
+            invoice_no VARCHAR(50),
+            stock_code VARCHAR(50),
+            description TEXT,
+            quantity INTEGER,
+            invoice_date TIMESTAMP,
+            unit_price NUMERIC,
+            customer_id VARCHAR(50),
+            country VARCHAR(100)
+        );
+        """,
+
+        # 3. DIM: Товары
+        """
+        CREATE TABLE IF NOT EXISTS star.dim_products (
+            stock_code VARCHAR(50) PRIMARY KEY,
+            description TEXT
+        );
+        """,
+
+        # 4. DIM: Клиенты
+        """
+        CREATE TABLE IF NOT EXISTS star.dim_customers (
+            customer_id VARCHAR(50) PRIMARY KEY,
+            country VARCHAR(100)
+        );
+        """,
+
+        # 5. FACT: Продажи
+        """
+        CREATE TABLE IF NOT EXISTS star.fact_sales (
+            sales_id SERIAL PRIMARY KEY,
+            invoice_no VARCHAR(50),
+            invoice_date TIMESTAMP,
+            customer_id VARCHAR(50),
+            stock_code VARCHAR(50),
+            quantity INTEGER,
+            unit_price NUMERIC,
+            total_amount NUMERIC,
+            -- Уникальный ключ для идемпотентности
+            UNIQUE(invoice_no, stock_code, customer_id) 
+        );
+        """
+    ]
+
+    try:
+        print("Подключение к базе данных...")
+        conn = psycopg2.connect(**conn_params)
+        cur = conn.cursor()
+
+        for command in commands:
+            cur.execute(command)
+
+        conn.commit()
+        cur.close()
+        conn.close()
+        print("SUCCESS: Star Schema tables created successfully!")
+
+    except Exception as e:
+        print(f"ERROR: Could not create tables. {e}")
+
+if __name__ == "__main__":
+    create_tables()