Polars pour Data Engineers

# Créer un fichier de test
import random
import csv
import os

os.makedirs("data", exist_ok=True)

# Générer 500K lignes
categories = ["Electronics", "Clothing", "Food", "Books", "Sports"]
n_rows = 500_000

with open("data/benchmark.csv", "w", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(["id", "category", "amount", "quantity"])
    for i in range(n_rows):
        writer.writerow([
            i,
            random.choice(categories),
            round(random.uniform(10, 1000), 2),
            random.randint(1, 100)
        ])

print(f"✅ Fichier créé : data/benchmark.csv ({n_rows:,} lignes)")

import pandas as pd
import time

# Benchmark Pandas
start = time.time()

df_pandas = pd.read_csv("data/benchmark.csv")
result_pandas = (
    df_pandas
    .groupby("category")
    .agg({"amount": "sum", "quantity": "mean"})
    .reset_index()
)

pandas_time = time.time() - start
print(f"🐼 Pandas : {pandas_time:.3f} secondes")
print(result_pandas)

import polars as pl
import time

# Benchmark Polars (Eager)
start = time.time()

df_polars = pl.read_csv("data/benchmark.csv")
result_polars = (
    df_polars
    .group_by("category")
    .agg(
        pl.col("amount").sum().alias("amount_sum"),
        pl.col("quantity").mean().alias("quantity_mean")
    )
)

polars_time = time.time() - start
print(f"🐻‍❄️ Polars : {polars_time:.3f} secondes")
print(f"⚡ Polars est {pandas_time/polars_time:.1f}x plus rapide !")
print(result_polars)

# Benchmark Polars (Lazy) - encore plus rapide !
start = time.time()

result_lazy = (
    pl.scan_csv("data/benchmark.csv")  # Lazy !
    .group_by("category")
    .agg(
        pl.col("amount").sum().alias("amount_sum"),
        pl.col("quantity").mean().alias("quantity_mean")
    )
    .collect()  # Exécution optimisée
)

lazy_time = time.time() - start
print(f"🚀 Polars Lazy : {lazy_time:.3f} secondes")
print(f"⚡ Polars Lazy est {pandas_time/lazy_time:.1f}x plus rapide que Pandas !")

import polars as pl

print(f"✅ Polars version : {pl.__version__}")

# Configuration de l'affichage
pl.Config.set_tbl_rows(10)           # Lignes affichées
pl.Config.set_tbl_cols(12)           # Colonnes affichées
pl.Config.set_fmt_str_lengths(50)    # Longueur des strings

# Voir le nombre de threads utilisés
print(f"🔧 Threads disponibles : {pl.thread_pool_size()}")

import polars as pl

# ============ EAGER (tout en mémoire) ============
df = pl.read_csv("data/benchmark.csv")
print("Eager - Type:", type(df))
print(df.head(3))

print("\n" + "="*50 + "\n")

# ============ LAZY (plan d'exécution) ============
lf = pl.scan_csv("data/benchmark.csv")
print("Lazy - Type:", type(lf))
print(lf)  # Affiche le plan, pas les données

# Créer plusieurs fichiers pour l'exemple
import os
os.makedirs("data/multi", exist_ok=True)

for i in range(3):
    pl.DataFrame({
        "id": range(i*100, (i+1)*100),
        "value": [i*10 + j for j in range(100)]
    }).write_csv(f"data/multi/file_{i}.csv")

print("✅ Fichiers créés")

# Lire plusieurs fichiers avec glob pattern
lf = pl.scan_csv("data/multi/*.csv")
print(f"\nNombre de lignes : {lf.collect().height}")

# Écriture
df = pl.DataFrame({
    "name": ["Alice", "Bob", "Charlie"],
    "age": [25, 30, 35],
    "city": ["Paris", "Lyon", "Marseille"]
})

# CSV
df.write_csv("data/output.csv")

# Parquet (recommandé pour la production)
df.write_parquet("data/output.parquet")

# JSON
df.write_json("data/output.json")

print("✅ Fichiers exportés")

# Vérifier avec Parquet
df_parquet = pl.read_parquet("data/output.parquet")
print(df_parquet)

import polars as pl

df = pl.DataFrame({
    "name": ["Alice", "Bob", "Charlie", "Diana"],
    "age": [25, 30, 35, 28],
    "salary": [50000, 60000, 75000, 55000],
    "department": ["IT", "HR", "IT", "Finance"]
})

print("DataFrame original :")
print(df)

# Sélectionner des colonnes avec expressions
print("\nSélection avec expressions :")
print(
    df.select(
        pl.col("name"),
        pl.col("salary") / 12,  # Salaire mensuel
    )
)

# Expressions conditionnelles : when/then/otherwise
print("Expressions conditionnelles :")
print(
    df.with_columns(
        pl.when(pl.col("age") >= 30)
          .then(pl.lit("Senior"))
          .otherwise(pl.lit("Junior"))
          .alias("level"),
        
        pl.when(pl.col("salary") > 60000)
          .then(pl.lit("High"))
          .when(pl.col("salary") > 50000)
          .then(pl.lit("Medium"))
          .otherwise(pl.lit("Low"))
          .alias("salary_band")
    )
)

# Chaînage d'expressions
print("Chaînage d'expressions :")
print(
    df.with_columns(
        # String operations
        pl.col("name").str.to_uppercase().alias("NAME_UPPER"),
        pl.col("name").str.len_chars().alias("name_length"),
        
        # Math operations
        (pl.col("salary") * 1.1).round(2).alias("salary_raised"),
    )
)

df = pl.read_csv("data/benchmark.csv")

# Sélection simple
print("Sélection simple :")
print(df.select("category", "amount").head(3))

# Sélection avec transformation
print("\nSélection avec transformation :")
print(
    df.select(
        pl.col("category"),
        (pl.col("amount") * pl.col("quantity")).alias("total")
    ).head(3)
)

# Sélection par type
print("\nColonnes numériques uniquement :")
print(df.select(pl.col(pl.Float64, pl.Int64)).head(3))

# Filtre simple
print("Filtre simple (amount > 500) :")
print(df.filter(pl.col("amount") > 500).head(3))

# Filtres multiples (AND)
print("\nFiltres multiples (AND) :")
print(
    df.filter(
        (pl.col("amount") > 500) & 
        (pl.col("category") == "Electronics")
    ).head(3)
)

# Filtres multiples (OR)
print("\nFiltres multiples (OR) :")
print(
    df.filter(
        (pl.col("category") == "Electronics") | 
        (pl.col("category") == "Books")
    ).head(3)
)

# Filtre avec is_in
print("\nFiltre avec is_in :")
print(
    df.filter(
        pl.col("category").is_in(["Electronics", "Books"])
    ).head(3)
)

print("Ajout de colonnes :")
result = df.with_columns(
    # Calcul
    (pl.col("amount") * pl.col("quantity")).alias("total"),
    
    # Valeur constante
    pl.lit("USD").alias("currency"),
    
    # Transformation de colonne existante
    pl.col("category").str.to_uppercase().alias("CATEGORY"),
    
    # Conditionnel
    pl.when(pl.col("amount") > 500)
      .then(pl.lit("High"))
      .otherwise(pl.lit("Low"))
      .alias("amount_level")
)

print(result.head(5))

print("GroupBy simple :")
print(
    df.group_by("category").agg(
        pl.col("amount").sum().alias("total_amount"),
        pl.col("amount").mean().alias("avg_amount"),
        pl.col("amount").max().alias("max_amount"),
        pl.len().alias("count")
    ).sort("total_amount", descending=True)
)

# Aggregations avancées
print("Aggregations avancées :")
print(
    df.group_by("category").agg(
        # Statistiques
        pl.col("amount").mean().alias("avg"),
        pl.col("amount").std().alias("std"),
        pl.col("amount").quantile(0.5).alias("median"),
        
        # Comptages conditionnels
        (pl.col("amount") > 500).sum().alias("high_amount_count"),
        
        # Premier/Dernier
        pl.col("amount").first().alias("first_amount"),
    )
)

# Tri
print("Tri décroissant :")
print(df.sort("amount", descending=True).head(3))

# Tri multiple
print("\nTri multiple :")
print(df.sort(["category", "amount"], descending=[True, False]).head(5))

# Renommer
print("\nRenommer :")
print(df.rename({"amount": "montant", "quantity": "quantite"}).head(2))

# Supprimer des colonnes
print("\nSupprimer colonnes :")
print(df.drop("id").head(2))

# Créer des DataFrames pour les joins
orders = pl.DataFrame({
    "order_id": [1, 2, 3, 4],
    "customer_id": [101, 102, 101, 103],
    "amount": [100, 200, 150, 300]
})

customers = pl.DataFrame({
    "customer_id": [101, 102, 104],
    "name": ["Alice", "Bob", "Diana"]
})

print("Orders:", orders)
print("\nCustomers:", customers)

# Inner join
print("\nInner Join :")
print(orders.join(customers, on="customer_id", how="inner"))

# Left join
print("\nLeft Join :")
print(orders.join(customers, on="customer_id", how="left"))

from datetime import datetime, date

df_dates = pl.DataFrame({
    "event": ["A", "B", "C", "D"],
    "timestamp": [
        datetime(2024, 1, 15, 10, 30),
        datetime(2024, 3, 20, 14, 45),
        datetime(2024, 6, 5, 9, 0),
        datetime(2024, 12, 25, 18, 30)
    ]
})

print("DataFrame avec dates :")
print(df_dates)

print("\nExtractions de dates :")
print(
    df_dates.with_columns(
        pl.col("timestamp").dt.year().alias("year"),
        pl.col("timestamp").dt.month().alias("month"),
        pl.col("timestamp").dt.day().alias("day"),
        pl.col("timestamp").dt.hour().alias("hour"),
        pl.col("timestamp").dt.weekday().alias("weekday"),
        pl.col("timestamp").dt.strftime("%Y-%m-%d").alias("date_str"),
    )
)

# Depuis un fichier (recommandé)
lf = pl.scan_csv("data/benchmark.csv")
print("Type:", type(lf))
print("\nLazyFrame (pas encore exécuté) :")
print(lf)

# Depuis un DataFrame existant
df = pl.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
lf = df.lazy()
print("Converti en LazyFrame:", type(lf))

# Pipeline complet en Lazy
pipeline = (
    pl.scan_csv("data/benchmark.csv")
    .filter(pl.col("amount") > 100)
    .with_columns(
        (pl.col("amount") * pl.col("quantity")).alias("total"),
        pl.col("category").str.to_uppercase().alias("CATEGORY")
    )
    .group_by("CATEGORY")
    .agg(
        pl.col("total").sum().alias("total_revenue"),
        pl.len().alias("transaction_count")
    )
    .sort("total_revenue", descending=True)
)

print("Pipeline défini (pas encore exécuté) :")
print(pipeline)
print("\n⚠️ Rien n'a été lu ou calculé !")

import time

start = time.time()
result = pipeline.collect()  # MAINTENANT ça s'exécute
print(f"⏱️ Temps d'exécution : {time.time() - start:.3f}s")
print("\nRésultat :")
print(result)

# Plan logique (ce que tu as écrit)
print("=== PLAN LOGIQUE ===")
print(pipeline.explain())

print("\n" + "="*60 + "\n")

# Plan optimisé (ce que Polars exécute réellement)
print("=== PLAN OPTIMISÉ ===")
print(pipeline.explain(optimized=True))

import pandas as pd
import polars as pl

# Créer un DataFrame Pandas
pandas_df = pd.DataFrame({
    "name": ["Alice", "Bob"],
    "age": [25, 30]
})

# Pandas → Polars
polars_df = pl.from_pandas(pandas_df)
print("Pandas → Polars :")
print(polars_df)

# Polars → Pandas
back_to_pandas = polars_df.to_pandas()
print("\nPolars → Pandas :")
print(back_to_pandas)

# Exemple de migration complète

# ============ VERSION PANDAS ============
# df = pd.read_csv("data.csv")
# df = df[df["amount"] > 100]
# df["total"] = df["amount"] * df["quantity"]
# result = df.groupby("category").agg({"total": "sum"}).reset_index()

# ============ VERSION POLARS (Eager) ============
result_eager = (
    pl.read_csv("data/benchmark.csv")
    .filter(pl.col("amount") > 100)
    .with_columns(
        (pl.col("amount") * pl.col("quantity")).alias("total")
    )
    .group_by("category")
    .agg(pl.col("total").sum())
)

# ============ VERSION POLARS (Lazy - recommandé) ============
result_lazy = (
    pl.scan_csv("data/benchmark.csv")
    .filter(pl.col("amount") > 100)
    .with_columns(
        (pl.col("amount") * pl.col("quantity")).alias("total")
    )
    .group_by("category")
    .agg(pl.col("total").sum())
    .collect()
)

print("Résultat :")
print(result_lazy)

# ❌ MAUVAIS : apply() ligne par ligne
# df.map_rows(lambda row: row[0] * 2)  # TRÈS LENT

# ✅ BON : expression native
df = pl.DataFrame({"x": [1, 2, 3]})
result = df.with_columns((pl.col("x") * 2).alias("x_doubled"))
print("✅ Expression native :")
print(result)

# ❌ MAUVAIS : read_csv sur plusieurs fichiers séparément
# dfs = [pl.read_csv(f) for f in files]  # Pas optimisé

# ✅ BON : scan_csv avec glob
lf = pl.scan_csv("data/multi/*.csv")
print("✅ Scan avec glob :")
print(lf.collect())

# Setup : créer les données de test
import polars as pl
import random
from datetime import datetime, timedelta
import os

os.makedirs("data/raw", exist_ok=True)
os.makedirs("data/processed", exist_ok=True)

categories = ["Electronics", "Clothing", "Food", "Books", "Sports"]
base_date = datetime(2024, 1, 1)

# Générer 3 fichiers CSV
for file_num in range(1, 4):
    n_rows = 10000
    data = {
        "transaction_id": range(file_num * 10000, file_num * 10000 + n_rows),
        "timestamp": [base_date + timedelta(days=random.randint(0, 365)) for _ in range(n_rows)],
        "category": [random.choice(categories) for _ in range(n_rows)],
        "amount": [round(random.uniform(-50, 1000), 2) for _ in range(n_rows)],  # Certains négatifs !
        "quantity": [random.randint(0, 100) for _ in range(n_rows)],  # Certains à 0 !
        "customer_id": [random.randint(1000, 9999) for _ in range(n_rows)]
    }
    df = pl.DataFrame(data)
    df.write_csv(f"data/raw/transactions_{file_num:02d}.csv")

print("✅ Données de test créées (3 fichiers x 10,000 lignes)")

import polars as pl
import time

print("🚀 Démarrage du pipeline ETL Polars...\n")
start = time.time()

# ============ PIPELINE LAZY ============
result = (
    # 1. EXTRACT : Lire tous les CSV avec glob pattern
    pl.scan_csv("data/raw/*.csv")
    
    # 2. CLEAN : Filtrer les données invalides
    .filter(
        (pl.col("amount") > 0) &           # Montants positifs
        (pl.col("quantity") > 0) &         # Quantités positives
        (pl.col("customer_id").is_not_null())  # Pas de null
    )
    
    # 3. TRANSFORM : Enrichir les données
    .with_columns(
        # Calculer le total
        (pl.col("amount") * pl.col("quantity")).alias("total_revenue"),
        
        # Extraire année et mois
        pl.col("timestamp").str.to_datetime().dt.year().alias("year"),
        pl.col("timestamp").str.to_datetime().dt.month().alias("month"),
        
        # Catégoriser les montants
        pl.when(pl.col("amount") > 500)
          .then(pl.lit("High"))
          .when(pl.col("amount") > 100)
          .then(pl.lit("Medium"))
          .otherwise(pl.lit("Low"))
          .alias("amount_tier"),
        
        # Uppercase category
        pl.col("category").str.to_uppercase().alias("category_upper")
    )
    
    # 4. AGGREGATE : Par catégorie et mois
    .group_by(["year", "month", "category_upper"])
    .agg(
        pl.col("total_revenue").sum().alias("total_revenue"),
        pl.col("total_revenue").mean().alias("avg_revenue"),
        pl.len().alias("transaction_count"),
        pl.col("customer_id").n_unique().alias("unique_customers"),
        (pl.col("amount_tier") == "High").sum().alias("high_value_count")
    )
    
    # 5. SORT
    .sort(["year", "month", "total_revenue"], descending=[False, False, True])
    
    # 6. EXECUTE
    .collect()
)

execution_time = time.time() - start
print(f"Pipeline exécuté en {execution_time:.3f} secondes")
print(f"Résultat : {result.height} lignes, {result.width} colonnes\n")

# Afficher un aperçu
print("Aperçu des résultats :")
print(result.head(10))

# 7. EXPORT en Parquet
result.write_parquet("data/processed/monthly_summary.parquet")
print("\n Résultat exporté : data/processed/monthly_summary.parquet")

# Vérifier le fichier Parquet
print("Lecture du fichier Parquet exporté :")
df_check = pl.read_parquet("data/processed/monthly_summary.parquet")
print(df_check.describe())

# Nettoyage des fichiers temporaires (optionnel)
import shutil
import os

# Décommenter pour nettoyer
# if os.path.exists("data"):
#     shutil.rmtree("data")
#     print("🧹 Dossier data/ supprimé")