import numpy as np

# Creation de tableaux performants
prix = np.array([29.99, 45.50, 12.00, 89.99, 34.50])
quantites = np.array([100, 50, 200, 25, 75])

# Operations vectorisees (pas de boucle Python!)
chiffre_affaires = prix * quantites
total = np.sum(chiffre_affaires)  # 12_361.25

# Statistiques instantanees
print(f"CA moyen par produit: {np.mean(chiffre_affaires):.2f}")
print(f"Ecart-type: {np.std(chiffre_affaires):.2f}")
print(f"Produit max: index {np.argmax(chiffre_affaires)}")

# Filtrage vectorise
produits_rentables = prix[chiffre_affaires > 2000]

import pandas as pd

# Lecture de multiples formats
df_csv = pd.read_csv("ventes.csv")
df_excel = pd.read_excel("rapport.xlsx", sheet_name="Q4")
df_json = pd.read_json("api_response.json")
df_parquet = pd.read_parquet("datalake/ventes.parquet")
df_sql = pd.read_sql("SELECT * FROM clients", engine)

# Exploration rapide
print(df_csv.shape)          # (1000000, 15)
print(df_csv.dtypes)         # Types de chaque colonne
print(df_csv.describe())     # Stats descriptives
print(df_csv.info(memory_usage='deep'))  # Usage memoire

import duckdb

# Requete directe sur un fichier Parquet (sans le charger en memoire!)
result = duckdb.sql("""
    SELECT
        region,
        product_category,
        SUM(amount) as total_sales,
        COUNT(DISTINCT customer_id) as unique_customers,
        AVG(amount) as avg_order_value
    FROM 'sales_2024/*.parquet'       -- Glob sur plusieurs fichiers!
    WHERE order_date >= '2024-01-01'
    GROUP BY region, product_category
    HAVING total_sales > 100000
    ORDER BY total_sales DESC
""").df()  # Convertit en pandas DataFrame

# Requete sur un fichier CSV distant
duckdb.sql("""
    SELECT * FROM read_csv_auto(
        'https://data.example.com/dataset.csv',
        header=true,
        dateformat='%Y-%m-%d'
    )
    LIMIT 100
""")

# Integration transparente avec pandas
import pandas as pd
df_clients = pd.read_csv("clients.csv")
df_commandes = pd.read_parquet("commandes.parquet")

# DuckDB peut joindre un DataFrame pandas avec un fichier Parquet!
result = duckdb.sql("""
    SELECT c.nom, c.ville, SUM(co.montant) as total
    FROM df_clients c                         -- DataFrame pandas
    JOIN df_commandes co ON c.id = co.client_id  -- Fichier Parquet
    GROUP BY c.nom, c.ville
    ORDER BY total DESC
    LIMIT 20
""").df()

import polars as pl

# Lecture de donnees
df = pl.read_parquet("ventes.parquet")

# API d'expressions (plus claire que pandas)
result = (
    df.lazy()                                    # Mode lazy
    .filter(pl.col("date") >= "2024-01-01")     # Filtre
    .group_by("region", "categorie")             # Groupby
    .agg([
        pl.col("montant").sum().alias("total_ventes"),
        pl.col("client_id").n_unique().alias("clients_uniques"),
        pl.col("montant").mean().alias("panier_moyen"),
        pl.col("montant").quantile(0.95).alias("p95_montant"),
    ])
    .sort("total_ventes", descending=True)
    .collect()                                   # Execute tout d'un coup!
)

# Polars optimise automatiquement le plan d'execution
# - Predicate pushdown (filtre avant groupby)
# - Projection pushdown (ne lit que les colonnes necessaires)
# - Parallelisation automatique sur tous les coeurs

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder \
    .appName("DataAnalysis") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

# Lecture depuis un data lake
df = spark.read.parquet("s3://datalake/ventes/year=2024/")

# API DataFrame similaire a pandas
result = (
    df.filter(F.col("status") == "completed")
      .groupBy("region", "category")
      .agg(
          F.sum("amount").alias("total"),
          F.countDistinct("customer_id").alias("unique_customers")
      )
      .orderBy(F.desc("total"))
)

# Ou en SQL pur
df.createOrReplaceTempView("ventes")
spark.sql("""
    SELECT region, category, SUM(amount) as total
    FROM ventes
    WHERE status = 'completed'
    GROUP BY region, category
    ORDER BY total DESC
""").show()

from sqlalchemy import create_engine

# Connexion a differentes bases
pg_engine = create_engine("postgresql://user:pass@host:5432/db")
mysql_engine = create_engine("mysql+pymysql://user:pass@host:3306/db")
sqlite_engine = create_engine("sqlite:///local.db")

# Integration pandas : le duo parfait
import pandas as pd
df = pd.read_sql("SELECT * FROM clients WHERE actif = true", pg_engine)
df_transforme.to_sql("clients_clean", pg_engine, if_exists="replace", index=False)

import pandas as pd
import numpy as np

# 1. Depuis un dictionnaire (le plus courant en dev)
df = pd.DataFrame({
    'nom': ['Alice', 'Bob', 'Charlie', 'Diana', 'Eve'],
    'age': [28, 34, 45, 31, 27],
    'ville': ['Paris', 'Lyon', 'Paris', 'Marseille', 'Lyon'],
    'salaire': [55000, 62000, 78000, 58000, 51000],
    'departement': ['Tech', 'Marketing', 'Tech', 'RH', 'Tech']
})

# 2. Depuis un fichier CSV (le plus courant en data)
df_ventes = pd.read_csv(
    "ventes.csv",
    sep=";",                      # Separateur (defaut: virgule)
    encoding="utf-8",             # Encodage
    parse_dates=["date_vente"],   # Colonnes date
    dtype={"code_postal": str},   # Forcer les types
    usecols=["date_vente", "montant", "client_id"],  # Colonnes utiles
    na_values=["N/A", "NULL", ""],  # Valeurs manquantes custom
    nrows=100000                  # Limiter les lignes (test)
)

# 3. Depuis une base de donnees SQL
from sqlalchemy import create_engine
engine = create_engine("postgresql://user:pass@localhost/ecommerce")
df_clients = pd.read_sql("""
    SELECT id, nom, email, date_inscription, ltv
    FROM clients
    WHERE actif = true
    ORDER BY date_inscription DESC
""", engine, parse_dates=["date_inscription"])

# 4. Depuis un fichier Parquet (recommande pour la data)
df_logs = pd.read_parquet(
    "logs_2024.parquet",
    columns=["timestamp", "event_type", "user_id"]  # Projection!
)

# 5. Depuis un fichier JSON (API responses)
df_api = pd.read_json("api_response.json", orient="records")

# 6. Depuis un fichier Excel
df_budget = pd.read_excel(
    "budget_2024.xlsx",
    sheet_name="Q4",
    skiprows=2,         # Sauter les lignes d'en-tete custom
    header=0            # Ligne d'en-tete apres skip
)

# Dimensions et structure
print(f"Shape: {df.shape}")          # (1000, 5)
print(f"Colonnes: {df.columns.tolist()}")
print(f"Types:\n{df.dtypes}")

# Premieres et dernieres lignes
df.head(10)
df.tail(5)
df.sample(5)   # 5 lignes aleatoires

# Statistiques descriptives
df.describe()                    # Colonnes numeriques
df.describe(include='object')    # Colonnes texte

# Informations memoire detaillees
df.info(memory_usage='deep')

# Valeurs manquantes
print(df.isnull().sum())                    # Par colonne
print(f"Total NaN: {df.isnull().sum().sum()}")
print(f"% complet: {df.notna().mean() * 100}")  # % par colonne

# Valeurs uniques
print(df['ville'].nunique())         # Nombre de valeurs uniques
print(df['ville'].value_counts())    # Distribution des valeurs

# Doublons
print(f"Doublons: {df.duplicated().sum()}")
print(f"Doublons sur nom+ville: {df.duplicated(subset=['nom','ville']).sum()}")

# ===== SELECTION PAR LABEL : .loc =====
# Une seule valeur
df.loc[0, 'nom']                    # 'Alice'

# Une colonne entiere (Series)
df.loc[:, 'nom']                    # Equivalent: df['nom']

# Plusieurs colonnes
df.loc[:, ['nom', 'salaire']]       # Equivalent: df[['nom', 'salaire']]

# Plage de lignes et colonnes (inclusif!)
df.loc[1:3, 'nom':'ville']          # Lignes 1,2,3 et colonnes nom, age, ville

# Avec condition booleenne (le plus utile!)
df.loc[df['age'] > 30, ['nom', 'salaire']]

# ===== SELECTION PAR POSITION : .iloc =====
# Premiere ligne, premiere colonne
df.iloc[0, 0]                       # 'Alice'

# 5 premieres lignes, 2 premieres colonnes
df.iloc[:5, :2]

# Lignes et colonnes specifiques
df.iloc[[0, 2, 4], [0, 3]]          # Alice et Eve, nom et salaire

# Derniere ligne
df.iloc[-1]

# ===== SELECTION BOOLEENNE (MASQUES) =====
# Condition simple
jeunes = df[df['age'] < 30]

# Conditions combinees (& = AND, | = OR, ~ = NOT)
tech_seniors = df[(df['departement'] == 'Tech') & (df['age'] > 30)]
paris_ou_lyon = df[df['ville'].isin(['Paris', 'Lyon'])]
pas_paris = df[~(df['ville'] == 'Paris')]

# Condition sur texte
contient_a = df[df['nom'].str.contains('a', case=False)]
commence_par = df[df['nom'].str.startswith('A')]

# Condition avec query() - plus lisible
tech_seniors2 = df.query("departement == 'Tech' and age > 30")

# ===== CREATION DE COLONNES =====
# Calcul simple
df['salaire_mensuel'] = df['salaire'] / 12

# Calcul conditionnel
df['seniorite'] = np.where(df['age'] >= 35, 'Senior', 'Junior')

# Conditions multiples avec np.select
conditions = [
    df['salaire'] < 40000,
    df['salaire'] < 60000,
    df['salaire'] < 80000,
    df['salaire'] >= 80000
]
choix = ['Bas', 'Moyen', 'Eleve', 'Tres eleve']
df['tranche_salaire'] = np.select(conditions, choix, default='Inconnu')

# Depuis une autre colonne (string operations)
df['prenom'] = df['nom'].str.split(' ').str[0]
df['email_domain'] = df['email'].str.extract(r'@(.+)$')
df['nom_upper'] = df['nom'].str.upper()

# ===== GESTION DES VALEURS MANQUANTES =====
# Detecter
df.isnull().sum()

# Supprimer les lignes avec NaN
df_clean = df.dropna()                      # Toutes colonnes
df_clean = df.dropna(subset=['salaire'])     # Colonnes specifiques

# Remplir les NaN
df['salaire'] = df['salaire'].fillna(df['salaire'].median())
df['ville'] = df['ville'].fillna('Inconnu')
df['date'] = df['date'].ffill()   # Forward fill (propager la valeur precedente)

# ===== RENOMMAGE ET SUPPRESSION =====
df = df.rename(columns={'nom': 'name', 'age': 'years'})
df = df.drop(columns=['colonne_inutile', 'temp'])

# ===== TRI =====
df = df.sort_values('salaire', ascending=False)
df = df.sort_values(['ville', 'salaire'], ascending=[True, False])

# ===== CONVERSION DE TYPES =====
df['date'] = pd.to_datetime(df['date'], format='%Y-%m-%d')
df['montant'] = pd.to_numeric(df['montant'], errors='coerce')  # NaN si erreur
df['code'] = df['code'].astype(str)

# ===== OPERATIONS SUR LES DATES =====
df['annee'] = df['date'].dt.year
df['mois'] = df['date'].dt.month
df['jour_semaine'] = df['date'].dt.day_name()
df['trimestre'] = df['date'].dt.quarter
df['jours_depuis'] = (pd.Timestamp.now() - df['date']).dt.days

# Pattern 1 : Remplacer apply() par des operations vectorisees
# MAUVAIS
df['full_name'] = df.apply(lambda r: f"{r['prenom']} {r['nom']}", axis=1)
# BON
df['full_name'] = df['prenom'] + ' ' + df['nom']

# Pattern 2 : Remplacer les boucles de calcul
# MAUVAIS
totaux = []
for _, row in df.iterrows():
    totaux.append(row['prix'] * row['quantite'] * (1 - row['remise']))
df['total'] = totaux
# BON
df['total'] = df['prix'] * df['quantite'] * (1 - df['remise'])

# Pattern 3 : Conditions multiples avec np.select
conditions = [
    (df['score'] >= 90),
    (df['score'] >= 70),
    (df['score'] >= 50),
]
labels = ['A', 'B', 'C']
df['grade'] = np.select(conditions, labels, default='F')

# Pattern 4 : Mapping de valeurs
mapping = {'Paris': 'IDF', 'Lyon': 'ARA', 'Marseille': 'PACA'}
df['region'] = df['ville'].map(mapping)

# Pattern 5 : Binning numerique
df['tranche_age'] = pd.cut(
    df['age'],
    bins=[0, 25, 35, 50, 100],
    labels=['Junior', 'Confirme', 'Senior', 'Expert']
)

import pandas as pd
import numpy as np

# Donnees exemple : ventes e-commerce
df = pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=1000, freq='h'),
    'region': np.random.choice(['IDF', 'PACA', 'ARA', 'NAQ'], 1000),
    'categorie': np.random.choice(['Electronique', 'Mode', 'Maison'], 1000),
    'montant': np.random.uniform(10, 500, 1000).round(2),
    'client_id': np.random.randint(1, 200, 1000),
    'quantite': np.random.randint(1, 10, 1000)
})

# ===== AGGREGATION SIMPLE =====
# Chiffre d'affaires par region
ca_region = df.groupby('region')['montant'].sum()

# Plusieurs metriques a la fois
stats = df.groupby('region')['montant'].agg(['sum', 'mean', 'median', 'std', 'count'])

# ===== AGGREGATION MULTIPLE (named agg) =====
rapport = df.groupby('region').agg(
    ca_total=('montant', 'sum'),
    panier_moyen=('montant', 'mean'),
    nb_commandes=('montant', 'count'),
    clients_uniques=('client_id', 'nunique'),
    montant_max=('montant', 'max'),
    quantite_totale=('quantite', 'sum')
).round(2)

# ===== GROUPBY MULTI-NIVEAUX =====
croise = df.groupby(['region', 'categorie']).agg(
    ca=('montant', 'sum'),
    nb=('montant', 'count')
).reset_index()  # Important pour avoir un DataFrame plat

# ===== AGGREGATION AVEC FONCTIONS CUSTOM =====
def coefficient_variation(x):
    return x.std() / x.mean() * 100

volatilite = df.groupby('region')['montant'].agg(
    moyenne='mean',
    ecart_type='std',
    cv=coefficient_variation
)

# ===== TRANSFORM (ajouter le resultat au DataFrame original) =====
# Ajouter le CA total de la region a chaque ligne
df['ca_region'] = df.groupby('region')['montant'].transform('sum')

# Pourcentage du CA regional
df['pct_region'] = df['montant'] / df.groupby('region')['montant'].transform('sum') * 100

# Z-score par region (normalisation)
df['zscore'] = df.groupby('region')['montant'].transform(
    lambda x: (x - x.mean()) / x.std()
)

# ===== FILTER (garder/exclure des groupes entiers) =====
# Regions avec plus de 200 commandes
regions_actives = df.groupby('region').filter(lambda g: len(g) > 200)

# Clients avec un panier moyen > 200
gros_clients = df.groupby('client_id').filter(
    lambda g: g['montant'].mean() > 200
)

# DataFrames a joindre
clients = pd.DataFrame({
    'client_id': [1, 2, 3, 4, 5],
    'nom': ['Alice', 'Bob', 'Charlie', 'Diana', 'Eve'],
    'ville': ['Paris', 'Lyon', 'Paris', 'Marseille', 'Lyon'],
    'segment': ['Gold', 'Silver', 'Gold', 'Bronze', 'Silver']
})

commandes = pd.DataFrame({
    'order_id': [101, 102, 103, 104, 105, 106],
    'client_id': [1, 2, 1, 3, 6, 2],      # client 6 n'existe pas!
    'montant': [150, 200, 75, 300, 50, 120],
    'date': pd.date_range('2024-01-01', periods=6, freq='W')
})

# INNER JOIN (par defaut) - uniquement les correspondances
inner = pd.merge(commandes, clients, on='client_id', how='inner')
# 5 lignes (client_id=6 exclu car pas dans clients)

# LEFT JOIN - tout le DataFrame de gauche
left = pd.merge(commandes, clients, on='client_id', how='left')
# 6 lignes, client_id=6 avec NaN pour nom/ville/segment

# RIGHT JOIN - tout le DataFrame de droite
right = pd.merge(commandes, clients, on='client_id', how='right')
# inclut Diana et Eve (pas de commandes) avec NaN pour order_id/montant

# FULL OUTER JOIN - tout des deux cotes
outer = pd.merge(commandes, clients, on='client_id', how='outer')
# Inclut client_id=6 ET Diana/Eve

# Jointure sur colonnes differentes
pd.merge(df1, df2, left_on='id_client', right_on='customer_id')

# Jointure sur plusieurs colonnes
pd.merge(df1, df2, on=['annee', 'mois', 'region'])

# Verifier les doublons de jointure (indicator)
result = pd.merge(commandes, clients, on='client_id', how='outer', indicator=True)
print(result['_merge'].value_counts())
# both          5  (dans les deux)
# left_only     1  (client_id=6, uniquement dans commandes)
# right_only    2  (Diana, Eve, uniquement dans clients)

# Pivot table simple : CA par region et categorie
pivot = df.pivot_table(
    values='montant',
    index='region',           # Lignes
    columns='categorie',      # Colonnes
    aggfunc='sum',            # Fonction d'agregation
    fill_value=0,             # Remplacer NaN par 0
    margins=True,             # Ajouter les totaux
    margins_name='Total'
)

# Pivot avec plusieurs metriques
pivot_multi = df.pivot_table(
    values=['montant', 'quantite'],
    index='region',
    columns='categorie',
    aggfunc={
        'montant': ['sum', 'mean'],
        'quantite': 'sum'
    }
)

# Pivot temporel (mensuel par region)
df['mois'] = df['date'].dt.to_period('M')
pivot_time = df.pivot_table(
    values='montant',
    index='mois',
    columns='region',
    aggfunc='sum'
)

# Crosstab : frequences croisees (raccourci)
freq = pd.crosstab(
    df['region'],
    df['categorie'],
    normalize='index'   # Pourcentages par ligne
).round(3) * 100

# Donnees temporelles triees
df_ts = df.sort_values('date').set_index('date')

# ===== ROLLING (fenetre glissante) =====
# Moyenne mobile sur 7 jours
df_ts['mm7'] = df_ts['montant'].rolling(window=7).mean()

# Moyenne mobile ponderee
df_ts['mm_exp'] = df_ts['montant'].ewm(span=7).mean()

# Rolling par groupe
df['mm7_region'] = (df.sort_values('date')
    .groupby('region')['montant']
    .rolling(7).mean()
    .reset_index(level=0, drop=True))

# ===== EXPANDING (cumul depuis le debut) =====
df_ts['cumul'] = df_ts['montant'].expanding().sum()
df_ts['max_historique'] = df_ts['montant'].expanding().max()

# ===== SHIFT (decalage temporel) =====
df_ts['montant_precedent'] = df_ts['montant'].shift(1)     # Jour precedent
df_ts['montant_suivant'] = df_ts['montant'].shift(-1)      # Jour suivant
df_ts['variation'] = df_ts['montant'].pct_change()          # % variation

# ===== RANK (classement) =====
df['rank_montant'] = df.groupby('region')['montant'].rank(
    method='dense', ascending=False
)

# Top 3 par region
top3 = df[df['rank_montant'] <= 3].sort_values(
    ['region', 'rank_montant']
)

def optimize_dataframe(df, verbose=True):
    """Reduit l'empreinte memoire d'un DataFrame."""
    start_mem = df.memory_usage(deep=True).sum() / 1024**2

    for col in df.columns:
        col_type = df[col].dtype

        if col_type == 'object':
            # Convertir en category si peu de valeurs uniques
            ratio = df[col].nunique() / len(df)
            if ratio < 0.5:  # Moins de 50% de valeurs uniques
                df[col] = df[col].astype('category')

        elif col_type in ['int64', 'int32']:
            # Downcast entiers
            col_min = df[col].min()
            col_max = df[col].max()
            if col_min >= 0:
                if col_max < 255:
                    df[col] = df[col].astype(np.uint8)
                elif col_max < 65535:
                    df[col] = df[col].astype(np.uint16)
                elif col_max < 4294967295:
                    df[col] = df[col].astype(np.uint32)
            else:
                if col_min > -128 and col_max < 127:
                    df[col] = df[col].astype(np.int8)
                elif col_min > -32768 and col_max < 32767:
                    df[col] = df[col].astype(np.int16)
                elif col_min > -2147483648 and col_max < 2147483647:
                    df[col] = df[col].astype(np.int32)

        elif col_type == 'float64':
            df[col] = pd.to_numeric(df[col], downcast='float')

    end_mem = df.memory_usage(deep=True).sum() / 1024**2
    if verbose:
        print(f"Memoire: {start_mem:.1f} MB -> {end_mem:.1f} MB "
              f"(-{(1-end_mem/start_mem)*100:.0f}%)")
    return df

# Exemple : 1M lignes
# Avant: 450 MB -> Apres: 85 MB (-81%)
df = optimize_dataframe(df)

from sqlalchemy import create_engine

# ===== CREATION D'ENGINE =====
# PostgreSQL
engine = create_engine(
    "postgresql+psycopg2://user:password@localhost:5432/ecommerce",
    pool_size=10,            # Connexions permanentes
    max_overflow=20,         # Connexions supplementaires en pic
    pool_timeout=30,         # Timeout si pas de connexion dispo (sec)
    pool_recycle=3600,       # Recycler apres 1h (evite timeout serveur)
    echo=False               # True pour voir le SQL genere (debug)
)

# MySQL
mysql_engine = create_engine(
    "mysql+pymysql://user:pass@host:3306/db?charset=utf8mb4"
)

# SQLite (fichier local, ideal pour tests)
sqlite_engine = create_engine("sqlite:///local_database.db")

# DuckDB (via SQLAlchemy!)
duckdb_engine = create_engine("duckdb:///analytics.duckdb")

# ===== UTILISATION BASIQUE =====
# Context manager (recommande - ferme automatiquement la connexion)
from sqlalchemy import text

with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM clients LIMIT 10"))
    for row in result:
        print(row.nom, row.email)

# Transaction explicite
with engine.begin() as conn:  # Auto-commit a la fin, rollback si exception
    conn.execute(text("UPDATE clients SET actif = false WHERE ltv = 0"))
    conn.execute(text("DELETE FROM sessions WHERE expire < NOW()"))

from sqlalchemy import MetaData, Table, Column, Integer, String, Float, DateTime
from sqlalchemy import select, insert, update, delete, func, and_, or_

# Reflexion : charger les tables existantes
metadata = MetaData()
metadata.reflect(bind=engine)
clients = metadata.tables['clients']
commandes = metadata.tables['commandes']

# SELECT avec conditions
stmt = (
    select(clients.c.nom, clients.c.email, clients.c.ville)
    .where(and_(
        clients.c.actif == True,
        clients.c.ltv > 1000
    ))
    .order_by(clients.c.ltv.desc())
    .limit(50)
)

with engine.connect() as conn:
    result = conn.execute(stmt)
    rows = result.fetchall()

# JOIN + GROUP BY
stmt = (
    select(
        clients.c.ville,
        func.count(commandes.c.id).label('nb_commandes'),
        func.sum(commandes.c.montant).label('ca_total'),
        func.avg(commandes.c.montant).label('panier_moyen')
    )
    .join(commandes, clients.c.id == commandes.c.client_id)
    .group_by(clients.c.ville)
    .having(func.count(commandes.c.id) > 10)
    .order_by(func.sum(commandes.c.montant).desc())
)

# INSERT
stmt = insert(clients).values(
    nom='Jean Dupont',
    email='jean@example.com',
    ville='Paris',
    actif=True
)

# Bulk INSERT (beaucoup plus rapide)
with engine.begin() as conn:
    conn.execute(
        insert(clients),
        [
            {'nom': 'Alice', 'email': 'alice@ex.com', 'ville': 'Lyon'},
            {'nom': 'Bob', 'email': 'bob@ex.com', 'ville': 'Paris'},
            # ... des milliers de lignes
        ]
    )

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy import ForeignKey
from datetime import datetime
from typing import List, Optional

# Base declarative (SQLAlchemy 2.0+)
class Base(DeclarativeBase):
    pass

class Client(Base):
    __tablename__ = 'clients'

    id: Mapped[int] = mapped_column(primary_key=True)
    nom: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255), unique=True)
    ville: Mapped[Optional[str]] = mapped_column(String(100))
    date_inscription: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    actif: Mapped[bool] = mapped_column(default=True)

    # Relation : un client a plusieurs commandes
    commandes: Mapped[List["Commande"]] = relationship(back_populates="client")

    def __repr__(self):
        return f"Client(id={self.id}, nom='{self.nom}')"

class Commande(Base):
    __tablename__ = 'commandes'

    id: Mapped[int] = mapped_column(primary_key=True)
    client_id: Mapped[int] = mapped_column(ForeignKey('clients.id'))
    montant: Mapped[float] = mapped_column(Float)
    date: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    statut: Mapped[str] = mapped_column(String(20), default='pending')

    # Relation inverse
    client: Mapped["Client"] = relationship(back_populates="commandes")

# Creation des tables
Base.metadata.create_all(engine)

# Session pour interagir
SessionLocal = sessionmaker(bind=engine)

# ===== OPERATIONS CRUD =====
with SessionLocal() as session:
    # CREATE
    nouveau_client = Client(nom="Marie Martin", email="marie@ex.com", ville="Paris")
    session.add(nouveau_client)
    session.commit()

    # READ
    client = session.query(Client).filter_by(email="marie@ex.com").first()
    print(client.nom)  # Marie Martin

    # UPDATE
    client.ville = "Lyon"
    session.commit()

    # Requetes avancees
    clients_paris = (
        session.query(Client)
        .filter(Client.ville == "Paris", Client.actif == True)
        .order_by(Client.nom)
        .limit(20)
        .all()
    )

import pandas as pd
from sqlalchemy import create_engine, text

engine = create_engine("postgresql://user:pass@localhost/ecommerce")

# ===== LIRE DEPUIS LA BASE =====
# Methode 1 : SQL brut (le plus courant et flexible)
df = pd.read_sql("""
    SELECT c.nom, c.ville, SUM(co.montant) as ca_total
    FROM clients c
    JOIN commandes co ON c.id = co.client_id
    WHERE co.date >= '2024-01-01'
    GROUP BY c.nom, c.ville
    HAVING SUM(co.montant) > 1000
    ORDER BY ca_total DESC
""", engine)

# Methode 2 : avec parametres (securise contre SQL injection!)
df = pd.read_sql(
    text("SELECT * FROM clients WHERE ville = :ville AND actif = :actif"),
    engine,
    params={"ville": "Paris", "actif": True}
)

# Methode 3 : lire une table entiere
df = pd.read_sql_table("clients", engine, columns=["nom", "email", "ville"])

# ===== ECRIRE DANS LA BASE =====
# Ecriture simple
df_clean.to_sql(
    "clients_clean",
    engine,
    if_exists="replace",    # 'fail', 'replace', 'append'
    index=False,
    dtype={                  # Types SQL explicites
        'nom': String(100),
        'montant': Float,
        'date': DateTime
    },
    chunksize=10000,         # Insertion par lots (memoire)
    method='multi'           # INSERT multi-valeurs (plus rapide)
)

# Ecriture performante avec COPY (PostgreSQL)
from io import StringIO

def to_sql_copy(df, table_name, engine):
    """Utilise COPY pour un insert ultra-rapide (PostgreSQL)."""
    buffer = StringIO()
    df.to_csv(buffer, index=False, header=False, sep='\t')
    buffer.seek(0)
    with engine.raw_connection() as conn:
        with conn.cursor() as cur:
            cur.copy_from(buffer, table_name, sep='\t', null='')
        conn.commit()

# 10x plus rapide que to_sql() pour les gros volumes
to_sql_copy(df_million_rows, 'staging_table', engine)

from pyspark.sql import SparkSession

# Creation de la SparkSession (point d'entree unique)
spark = SparkSession.builder \
    .appName("DataArchitectTraining") \
    .master("local[*]") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.sql.parquet.compression.codec", "snappy") \
    .getOrCreate()

# ===== LECTURE DE DONNEES =====
# Parquet (format recommande pour Spark)
df = spark.read.parquet("s3://datalake/ventes/year=2024/")

# CSV
df_csv = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .option("sep", ";") \
    .csv("hdfs:///data/exports/ventes.csv")

# CSV avec schema explicite (recommande - evite inferSchema qui est lent)
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, \
    FloatType, TimestampType

schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("customer_id", IntegerType(), True),
    StructField("amount", FloatType(), True),
    StructField("order_date", TimestampType(), True),
    StructField("region", StringType(), True),
    StructField("category", StringType(), True)
])

df_typed = spark.read.schema(schema).csv("data/ventes.csv", header=True)

# JSON
df_json = spark.read.json("s3://bucket/events/*.json")

# Delta Lake
df_delta = spark.read.format("delta").load("s3://datalake/delta/clients")

# Base de donnees via JDBC
df_pg = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://host:5432/db") \
    .option("dbtable", "clients") \
    .option("user", "user") \
    .option("password", "pass") \
    .option("numPartitions", 10) \
    .option("partitionColumn", "id") \
    .option("lowerBound", 1) \
    .option("upperBound", 1000000) \
    .load()

from pyspark.sql import functions as F

# ===== TRANSFORMATIONS (rien ne s'execute encore!) =====
df_filtered = df.filter(F.col("amount") > 100)
df_selected = df_filtered.select("customer_id", "amount", "region")
df_grouped = df_selected.groupBy("region").agg(
    F.sum("amount").alias("total_sales"),
    F.count("*").alias("order_count"),
    F.avg("amount").alias("avg_amount"),
    F.countDistinct("customer_id").alias("unique_customers")
)
df_sorted = df_grouped.orderBy(F.desc("total_sales"))

# A ce stade, RIEN n'a ete execute! Spark a juste construit le plan.
# Vous pouvez verifier le plan :
df_sorted.explain(True)  # Affiche le plan logique et physique

# ===== ACTIONS (declenchent l'execution!) =====
df_sorted.show(20)                    # Affiche les 20 premieres lignes
total = df.count()                     # Compte les lignes
result = df_sorted.collect()           # Ramene TOUT au Driver (attention!)
first_row = df_sorted.first()          # Premiere ligne
pandas_df = df_sorted.toPandas()       # Convertit en pandas (attention!)

# ===== API DataFrame COMPLETE =====
result = (
    df
    # Filtrage
    .filter(F.col("order_date") >= "2024-01-01")
    .filter(F.col("status").isin(["completed", "shipped"]))

    # Nouvelles colonnes
    .withColumn("year", F.year("order_date"))
    .withColumn("month", F.month("order_date"))
    .withColumn("amount_eur", F.col("amount") * F.lit(0.92))
    .withColumn("is_premium", F.when(F.col("amount") > 500, True).otherwise(False))

    # Colonnes string
    .withColumn("region_upper", F.upper(F.col("region")))
    .withColumn("email_domain", F.regexp_extract("email", r"@(.+)$", 1))

    # Aggregation
    .groupBy("year", "month", "region")
    .agg(
        F.sum("amount_eur").alias("revenue"),
        F.count("*").alias("orders"),
        F.avg("amount_eur").alias("avg_order"),
        F.percentile_approx("amount_eur", 0.95).alias("p95_amount")
    )

    # Tri
    .orderBy("year", "month", F.desc("revenue"))
)

# Enregistrer un DataFrame comme vue temporaire
df.createOrReplaceTempView("orders")

# Requete SQL standard
result = spark.sql("""
    WITH monthly_stats AS (
        SELECT
            region,
            DATE_TRUNC('month', order_date) as month,
            SUM(amount) as revenue,
            COUNT(*) as order_count,
            COUNT(DISTINCT customer_id) as unique_customers
        FROM orders
        WHERE status = 'completed'
          AND order_date >= '2024-01-01'
        GROUP BY region, DATE_TRUNC('month', order_date)
    ),
    ranked AS (
        SELECT *,
            ROW_NUMBER() OVER (PARTITION BY month ORDER BY revenue DESC) as rank,
            LAG(revenue) OVER (PARTITION BY region ORDER BY month) as prev_revenue
        FROM monthly_stats
    )
    SELECT
        region, month, revenue, order_count, unique_customers,
        rank,
        ROUND((revenue - prev_revenue) / prev_revenue * 100, 1) as growth_pct
    FROM ranked
    WHERE rank <= 5
    ORDER BY month DESC, rank
""")

result.show(50, truncate=False)

# ===== 1. BROADCAST JOIN (petite table jointe a une grande) =====
from pyspark.sql.functions import broadcast

# Table de reference petite (< 100 MB)
dim_regions = spark.read.parquet("dim_regions.parquet")  # 1000 lignes

# MAUVAIS : shuffle join (redistribue les 100M lignes)
result = big_df.join(dim_regions, "region_id")

# BON : broadcast join (envoie la petite table a tous les executors)
result = big_df.join(broadcast(dim_regions), "region_id")
# 10-100x plus rapide car evite le shuffle!

# ===== 2. REPARTITIONNEMENT =====
# Trop de petites partitions (slow)
df.rdd.getNumPartitions()  # 10000 partitions de 1 MB

# Reduire le nombre de partitions (coalesce, sans shuffle)
df = df.coalesce(200)  # 200 partitions de 50 MB

# Repartitionner par cle (optimise les joins/groupby suivants)
df = df.repartition(200, "region")

# ===== 3. CACHING =====
# Si un DataFrame est utilise plusieurs fois
df_clients = spark.read.parquet("clients/").filter(F.col("actif"))
df_clients.cache()  # Stocke en memoire apres la premiere action
# ou
df_clients.persist(StorageLevel.MEMORY_AND_DISK)  # Spillover sur disque

# Utilise plusieurs fois sans relire le Parquet
rapport1 = df_clients.groupBy("region").count()
rapport2 = df_clients.groupBy("segment").agg(F.sum("ltv"))

df_clients.unpersist()  # Liberer la memoire quand fini

# ===== 4. PREDICATE PUSHDOWN =====
# Spark pousse les filtres vers la source (Parquet/JDBC)
# Parquet : ne lit que les colonnes et row groups necessaires
df = spark.read.parquet("big_table/") \
    .select("col1", "col2") \     # Projection pushdown
    .filter(F.col("date") >= "2024-01-01")  # Predicate pushdown

# ===== 5. AQE (Adaptive Query Execution) =====
# Active par defaut depuis Spark 3.2
spark.conf.set("spark.sql.adaptive.enabled", "true")
# AQE optimise le plan PENDANT l'execution :
# - Fusionne les petites partitions apres shuffle
# - Convertit sort-merge join en broadcast join si une side est petite
# - Gere le data skew automatiquement

nom,age,ville,salaire
Alice,28,Paris,55000
Bob,34,Lyon,62000
"Charlie ""Chuck""",45,Paris,78000

{"nom": "Alice", "age": 28, "ville": "Paris", "salaire": 55000}
{"nom": "Bob", "age": 34, "ville": "Lyon", "salaire": 62000}
{"nom": "Charlie", "age": 45, "ville": "Paris", "salaire": 78000, "bonus": 5000}

import pandas as pd
import pyarrow.parquet as pq

# Ecriture Parquet avec compression
df.to_parquet(
    "ventes.parquet",
    engine="pyarrow",
    compression="snappy",     # Alternatives: gzip, zstd, lz4
    index=False
)

# Ecriture partitionnee (essentiel pour les data lakes!)
df.to_parquet(
    "datalake/ventes/",
    partition_cols=["year", "month"],   # Cree des sous-dossiers
    compression="snappy"
)
# Resultat :
# datalake/ventes/year=2024/month=01/part-0.parquet
# datalake/ventes/year=2024/month=02/part-0.parquet
# ...

# Lecture avec projection pushdown (ne lit que les colonnes demandees)
df = pd.read_parquet(
    "ventes.parquet",
    columns=["date", "montant", "region"],   # Projection pushdown!
    filters=[("year", "=", 2024)]             # Predicate pushdown!
)

# Lecture des metadata sans charger les donnees
parquet_file = pq.read_metadata("ventes.parquet")
print(f"Lignes: {parquet_file.num_rows}")
print(f"Colonnes: {parquet_file.num_columns}")
print(f"Taille: {parquet_file.serialized_size / 1024**2:.1f} MB")
print(f"Row groups: {parquet_file.num_row_groups}")

import pyarrow as pa
import pyarrow.parquet as pq

# Conversion pandas <-> Arrow (quasi-instantane)
arrow_table = pa.Table.from_pandas(df)
df_back = arrow_table.to_pandas()

# Lecture Parquet via Arrow (plus rapide que pd.read_parquet)
table = pq.read_table("ventes.parquet", columns=["date", "montant"])
df = table.to_pandas()

# Arrow est utilise par :
# - Polars (backend natif)
# - DuckDB (echange avec pandas)
# - PySpark (echange Python <-> JVM depuis Spark 3.0)
# - Pandas 2.0 (backend optionnel: pd.ArrowDtype)

# Pandas avec backend Arrow (plus rapide et moins de memoire)
df = pd.read_parquet("data.parquet", dtype_backend="pyarrow")
print(df.dtypes)  # int64[pyarrow], string[pyarrow], etc.