curiositech/windags-skills

Optimiseur de Traitement par Lots

Optimisez les charges de travail de traitement par lots avec Spark, Polars, DuckDB et pandas, en mettant l'accent sur l'efficacité mémoire, le parallélisme et la réduction des coûts.

Déclencheurs d'Activation

Activer sur : "traitement par lots", "optimisation Spark", "Polars", "DuckDB", "performance pandas", "data frame", "optimisation shuffle", "partition skew", "optimisation mémoire", "dépassement mémoire"

PAS pour : Streaming temps réel → streaming-pipeline-architect | Optimisation SQL d'entrepôt → data-warehouse-optimizer | Orchestration de pipeline → airflow-dag-orchestrator

Démarrage Rapide

1. Choisir le bon outil — DuckDB pour l'analyse mono-nœud, Polars pour les DataFrames, Spark pour le distribué
2. Profiler d'abord — identifier les goulots d'étranglement (shuffle, skew, mémoire) avant d'optimiser
3. Réduire les données tôt — filtrer et sélectionner les colonnes le plus tôt possible dans le pipeline
4. Éviter les shuffles — diffuser les petites tables, pré-partitionner les données, utiliser les jointures côté map
5. Dimensionner correctement les ressources — adapter la mémoire et les cœurs de l'exécuteur à la taille réelle des données

Capacités de Base

Domaine	Technologies
Distribué	Apache Spark 3.5+, Dask, Ray
Mono-nœud	DuckDB 1.1+, Polars 1.x, pandas 2.2+
Formats de fichiers	Parquet, Arrow IPC, Delta Lake, Iceberg
Optimisation	AQE (Spark), évaluation paresseuse (Polars), scans columnaires
Cloud	Databricks, EMR, Dataproc, Spark sans serveur

Motifs d'Architecture

Arbre de Décision de Sélection d'Outil

Taille des données ?
  ├─ < 10 Go     → DuckDB (SQL) ou Polars (DataFrame)
  │                 Machine unique, zéro configuration, itération la plus rapide
  │
  ├─ 10-100 Go   → Polars (lazy) ou DuckDB (débordement sur disque)
  │                 Toujours mono-machine avec débordement sur disque
  │
  └─ > 100 Go    → Spark (distribué)
                    Cluster multi-nœud, jointures basées sur shuffle

Complexité ?
  ├─ Centré SQL  → DuckDB (moteur SQL le plus rapide pour l'analyse)
  ├─ DataFrame   → Polars (10x plus rapide que pandas, évaluation paresseuse)
  └─ ML complexe → Spark + MLlib ou Spark + Ray

Motifs d'Optimisation Spark

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

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

# BON : diffuser une petite table de dimension (< 100 Mo)
from pyspark.sql.functions import broadcast
result = large_df.join(broadcast(small_dim_df), "key")

# BON : pousser les prédicats — filtrer avant la jointure
orders = spark.read.parquet("s3://data/orders/") \
    .filter(F.col("order_date") >= "2026-01-01") \
    .select("order_id", "customer_id", "amount")  # élagage des colonnes

# MAUVAIS : collect() sur un grand jeu de données — provoque OOM sur le driver
# all_data = large_df.collect()  # NE JAMAIS faire ceci

# BON : écrire une sortie partitionnée
result.repartition(200) \
    .write.mode("overwrite") \
    .partitionBy("order_date") \
    .parquet("s3://output/results/")

Évaluation Paresseuse Polars

import polars as pl

# Mode paresseux : construit un plan de requête, optimise, puis exécute
result = (
    pl.scan_parquet("data/orders/*.parquet")  # scan paresseux
    .filter(pl.col("order_date") >= "2026-01-01")
    .join(
        pl.scan_parquet("data/customers/*.parquet"),
        on="customer_id",
        how="inner"
    )
    .group_by("region")
    .agg([
        pl.col("amount").sum().alias("total_revenue"),
        pl.col("order_id").n_unique().alias("order_count"),
    ])
    .sort("total_revenue", descending=True)
    .collect()  # exécute le plan optimisé
)

# Polars optimise : prédicat pushdown, projection pushdown,
# réordonnancement des jointures — tout automatiquement via l'évaluation paresseuse

Anti-Patrons

1. pandas pour >5 Go — pandas charge tout en mémoire ; utiliser Polars (paresseux) ou DuckDB pour les données moyennes, Spark pour les grandes
2. Collect vers le driver — df.collect() ou df.toPandas() sur de grands DataFrames Spark provoque OOM ; agréger d'abord
3. Ignorer le partition skew — une partition avec 10x plus de données que les autres bloque l'ensemble du travail ; utiliser AQE ou le salage
4. Lire toutes les colonnes — toujours sélectionner uniquement les colonnes nécessaires ; le format colonnaire Parquet ignore totalement les colonnes inutilisées
5. Fichiers de sortie minuscules — trop de petits fichiers de sortie (< 128 Mo) ralentissent les lectures en aval ; coalescer avant d'écrire

Checklist Qualité

• L'outil correspond à la taille des données (DuckDB/Polars < 100 Go, Spark > 100 Go)
• Les colonnes sont élaguées tôt (sélectionner uniquement ce qui est nécessaire)
• Les filtres sont poussés au niveau du scan (prédicat pushdown)
• Les petites tables sont diffusées dans les jointures (< 100 Mo)
• AQE Spark activé (exécution de requête adaptative)
• Pas de collect() sur les grands jeux de données (agréger avant de collecter)
• Les fichiers de sortie sont dimensionnés entre 128 Mo et 1 Go (coalescer/repartitionner avant écriture)
• La skew de partition est surveillée et atténuée (salage ou AQE)
• Le travail est profilé : étapes Spark UI, .explain() de Polars, EXPLAIN ANALYZE de DuckDB
• La mémoire est dimensionnée correctement : mémoire exécuteur >= 2x la plus grande partition