Pipeline

import pandas as pd
from sqlalchemy import create_engine
import numpy as np

def execute_etl_pipeline(sales_csv_path, sql_conn_string):
    # 1. EXTRACT: Lettura massiva con Types pre-ottimizzati per risparmiare RAM
    dtypes = {'user_id': 'int32', 'total': 'float32', 'status': 'category'}
    df_sales = pd.read_csv(sales_csv_path, dtype=dtypes, engine='c')
    
    engine = create_engine(sql_conn_string)
    df_users = pd.read_sql("SELECT id, country, acqusition_channel FROM users", con=engine)
    
    # 2. TRANSFORM: Data Cleansing & Business Logic
    # Drop ordini annullati e valori Null
    df_clean = df_sales[df_sales['status'] == 'Shipped'].dropna(subset=['total'])
    
    # Riconciliazione Numerica e Normalizzazione stringhe
    df_clean['total'] = df_clean['total'].apply(lambda x: np.round(abs(x), 2))
    
    # Table JOINs 
    df_master = pd.merge(df_clean, df_users, left_on='user_id', right_on='id', how='left')
    
    # Creazione Aggregati Dimensionali (Data Warehouse)
    df_cube = df_master.groupby(['country', 'acqusition_channel']).agg({
        'total': ['sum', 'mean'],
        'user_id': 'count'
    }).reset_index()
    df_cube.columns = ['Paese', 'Sorgente', 'Vendite_Totali', 'Scontrino_Medio', 'Num_Ordini']
    
    # 3. LOAD: Scrittura in Bulk sul Data Warehouse (es. Snowflake o BigQuery)
    df_cube.to_sql('vw_analytics_sales_cube', con=engine, if_exists='replace', index=False, chunksize=10000)
    
    return f"Pipeline completata. {len(df_clean)} records validati e trasferiti."