Cas pratiques Apache Spark

Contexte Une équipe analytique souhaite compter le nombre de commandes par pays à partir d'un fichier de commandes. Le Problème Le calcul est actuellement réalisé en Python natif avec une boucle, ce qui est inefficace sur un million de lignes. Contraintes Utiliser groupBy() et count() de PySpark. Trier les résultats par nombre de commandes décroissant. Aucune librairie externe. Livrable attendu Un script PySpark calculant le nombre de commandes par pays et affichant le résultat trié.

Contexte : Une entreprise souhaite calculer la moyenne des ventes par catégorie produit. Le Problème : Aucune agrégation n’a encore été implémentée dans Spark. Contraintes : Utiliser groupBy et agg avec la fonction avg. Conserver une approche DataFrame. Livrable attendu : Fournir un exemple de transformation Spark permettant de calculer la moyenne des ventes par catégorie.

Contexte : Un DataFrame est utilisé plusieurs fois dans le pipeline. Le Problème : Chaque action déclenche un recalcul complet du DAG. Extrait actuel : df.filter(...).count() df.filter(...).show() Aucun cache n’est défini. Contraintes : Utiliser cache() ou persist() de manière appropriée. Éviter la consommation mémoire excessive. Livrable attendu : Corriger le pipeline en introduisant un cache stratégique améliorant les performances.

Contexte : Un développeur tente d’appliquer une transformation sur une colonne qui n’existe pas dans le DataFrame. Le Problème : L’exécution échoue avec une erreur AnalysisException. Extrait actuel : df.select("non_existing_column") La colonne n’est pas présente dans le schéma. Contraintes : Vérifier le schéma avant transformation. Corriger le nom ou ajuster la logique. Livrable attendu : Proposer une correction permettant d’éviter l’erreur et garantir la robustesse du pipeline.

Contexte : Un fichier CSV est chargé dans Spark sans schéma explicite. Le Problème : Certaines colonnes numériques sont interprétées comme STRING, provoquant des erreurs lors des calculs. Extrait actuel : spark.read.option("header", true).csv("sales.csv") Aucun schéma n’est défini. Contraintes : Définir explicitement un schéma StructType adapté. Garantir la compatibilité avec les calculs numériques. Livrable attendu : Proposer une version corrigée du chargement avec un schéma explicite permettant des opérations arithmétiques fiables.

Contexte Un ingénieur data charge un fichier CSV de températures. Les colonnes numériques sont inférées comme StringType, provoquant des erreurs lors des calculs. Le Problème L'inférence automatique de schéma échoue sur certains fichiers mal formatés. Code actuel df = spark.read.csv("/data/temperatures.csv", header=True) df.select(avg("temperature")).show() Contraintes Définir un schéma explicite avec StructType et StructField. Forcer le type DoubleType sur la colonne temperature. Livrable attendu Un script PySpark définissant un schéma explicite et calculant correctement la moyenne de température.

Contexte : Un développeur utilise collect() pour récupérer toutes les données dans le driver. Le Problème : Sur un gros dataset, l’application plante avec une erreur OutOfMemory. Extrait actuel : val data = df.collect() Cette instruction ramène toutes les données en mémoire locale. Contraintes : Éviter collect() sur de grands volumes. Proposer une alternative distribuée. Livrable attendu : Corriger le code afin d’éviter le dépassement mémoire et conserver un traitement distribué efficace.

Contexte : Deux DataFrames sont joints pour enrichir des transactions avec des informations produits. Le Problème : La jointure actuelle produit un nombre inattendu de lignes. Extrait actuel : df1.join(df2) Aucune condition de jointure n’est spécifiée. Contraintes : Définir explicitement la clé de jointure. Éviter les cross joins involontaires. Livrable attendu : Proposer une jointure correcte basée sur une clé commune permettant d’obtenir un résultat cohérent.

Contexte : Votre entreprise reçoit chaque jour un fichier CSV contenant les transactions réalisées sur la plateforme. Le Problème : Aucun traitement distribué n’est encore en place pour agréger rapidement les ventes quotidiennes. Contraintes : Utiliser Apache Spark pour lire un fichier CSV. Calculer le total des ventes par jour. Utiliser les DataFrames API. Éviter toute logique locale non distribuée. Livrable attendu : Écrire un job Spark simple permettant de charger les données et produire une agrégation par date avec un affichage du résultat.

Contexte Une équipe data souhaite démarrer un traitement analytique sur un fichier de ventes au format CSV. Le Problème Le script Python actuel utilise pandas pour lire le fichier, ce qui ne passe pas à l'échelle sur des volumes importants. Code actuel import pandas as pd df = pd.read_csv("/data/sales.csv") print(df.head()) Contraintes Remplacer pandas par PySpark. Créer une SparkSession correctement configurée. Lire le fichier CSV avec inférence de schéma activée. Livrable attendu Un script PySpark fonctionnel créant une SparkSession, lisant le CSV et affichant les 5 premières lignes via show().

Contexte : Les données transformées doivent être stockées pour des analyses ultérieures. Le Problème : Aucun mécanisme d’écriture optimisé n’est encore en place. Contraintes : Écrire le DataFrame au format Parquet. Activer le mode overwrite si nécessaire. Optimiser la compression par défaut. Livrable attendu : Fournir un exemple de code Spark permettant d’écrire efficacement les données au format Parquet.

Contexte Un pipeline de données génère un DataFrame de résultats qui doit être persisté pour une consommation ultérieure. Le Problème Les résultats sont actuellement sauvegardés en CSV, format peu efficace pour les requêtes analytiques. Contraintes Écrire le DataFrame en format Parquet avec compression snappy. Utiliser le mode overwrite. Partitionner par la colonne pays. Livrable attendu Un script PySpark sauvegardant correctement un DataFrame en Parquet avec partitionnement et compression.

Contexte Un pipeline de données charge un fichier de transactions bancaires et doit extraire uniquement les transactions supérieures à 1000€. Le Problème Le script charge toutes les colonnes et toutes les lignes sans filtrage, générant un volume inutile en mémoire. Contraintes Utiliser les méthodes select() et filter() de l'API DataFrame. Ne conserver que les colonnes id, montant et date. Filtrer les lignes où montant > 1000. Livrable attendu Un script PySpark appliquant correctement sélection de colonnes et filtrage conditionnel sur le DataFrame.

Contexte Un DataFrame de produits contient des valeurs nulles dans les colonnes prix et categorie, bloquant les calculs d'agrégation. Le Problème Les fonctions d'agrégation retournent null dès qu'une valeur manquante est présente. Contraintes Remplacer les nulls de prix par 0.0 avec fillna(). Supprimer les lignes où categorie est null avec dropna(). Aucune librairie externe. Livrable attendu Un script PySpark gérant les nulls de manière appropriée selon la nature de chaque colonne.

Contexte Deux fichiers existent : un fichier de commandes et un fichier de clients. L'équipe souhaite enrichir les commandes avec les informations clients. Le Problème Le script actuel utilise une boucle Python pour associer chaque commande à son client, ce qui est impraticable à grande échelle. Contraintes Utiliser join() de PySpark avec une jointure de type inner sur la colonne client_id. Sélectionner uniquement les colonnes utiles du résultat. Livrable attendu Un script PySpark réalisant une jointure correcte entre les deux DataFrames et affichant le résultat enrichi.

Contexte : Une application collecte des données événementielles au format JSON. Le Problème : Les données doivent être chargées dans Spark pour traitement distribué. Contraintes : Utiliser spark.read.json pour importer les données. Vérifier le schéma généré. Préparer les données pour une future transformation. Livrable attendu : Écrire un exemple de code Spark permettant de lire un fichier JSON et afficher son schéma.

Contexte : Une équipe souhaite analyser uniquement les transactions supérieures à 100 unités monétaires. Le Problème : Le filtrage n’est pas encore implémenté dans le pipeline Spark. Contraintes : Utiliser l’API DataFrame pour filtrer les lignes. Éviter toute transformation en RDD. Conserver les performances distribuées. Livrable attendu : Écrire un extrait de code Spark permettant de filtrer efficacement les transactions selon le montant.

Contexte Un fichier source contient des colonnes mal nommées (espaces, majuscules) et des types incorrects qui bloquent les transformations suivantes. Le Problème Les étapes suivantes du pipeline échouent à cause de noms de colonnes incompatibles et de types StringType. Contraintes Utiliser withColumnRenamed() pour normaliser les noms. Utiliser withColumn() et cast() pour corriger les types. Aucune modification du fichier source. Livrable attendu Un script PySpark normalisant noms et types de colonnes pour préparer le DataFrame aux transformations suivantes.

Contexte Un fichier de contacts clients contient des lignes dupliquées issues de plusieurs sources d'import. Le Problème Les analyses produisent des résultats faussés à cause des doublons non supprimés. Contraintes Utiliser dropDuplicates() sur les colonnes email et telephone. Afficher le nombre de lignes avant et après déduplication. Livrable attendu Un script PySpark appliquant une déduplication ciblée et affichant les comptages comparatifs.

Contexte Une équipe habituée au SQL souhaite interroger un DataFrame Spark avec des requêtes SQL standard. Le Problème L'équipe ne maîtrise pas l'API DataFrame et préfère écrire des requêtes SQL pour les transformations. Contraintes Enregistrer le DataFrame comme vue temporaire avec createOrReplaceTempView(). Exécuter une requête SQL via spark.sql() pour calculer le chiffre d'affaires total par mois. Livrable attendu Un script PySpark démontrant l'utilisation de Spark SQL sur un DataFrame enregistré comme vue temporaire.

Contexte : Une petite table de référence est jointe à une table volumineuse. Le Problème : Spark n’utilise pas automatiquement le broadcast join, entraînant un shuffle inutile. Extrait actuel : df_large.join(df_small, "id") La table df_small est suffisamment petite pour être diffusée. Contraintes : Forcer un broadcast join si pertinent. Vérifier que la taille reste raisonnable. Livrable attendu : Proposer une version optimisée de la jointure utilisant broadcast pour réduire le shuffle.

Contexte Un job Spark de traitement de logs échoue régulièrement avec des erreurs OutOfMemoryError sur les executors. Le Problème Le job effectue plusieurs collect() sur des DataFrames volumineux et utilise des broadcast joins non maîtrisés. Code actuel result = large_df.join(another_large_df, "id").collect() for row in result: process(row) Contraintes Remplacer collect() par des actions adaptées. Utiliser broadcast() uniquement sur les petits DataFrames. Traiter les données sans ramener l'ensemble en mémoire driver. Livrable attendu Un script PySpark corrigé évitant les collect() abusifs et utilisant correctement les broadcast joins.

Contexte : Un DataFrame est persistant en mémoire mais provoque des erreurs liées au manque de RAM. Le Problème : persist() est utilisé sans niveau de stockage adapté. Extrait actuel : df.persist() Aucune stratégie MEMORY_AND_DISK n’est définie. Contraintes : Choisir un StorageLevel adapté. Éviter la saturation mémoire. Maintenir de bonnes performances. Livrable attendu : Proposer une configuration de persistance adaptée garantissant stabilité et performance.

Contexte : Une agrégation groupBy sur un dataset volumineux provoque un shuffle massif. Le Problème : Le plan d’exécution montre une redistribution excessive des données. Extrait actuel : df.groupBy("region").count() La colonne region contient peu de valeurs distinctes. Contraintes : Optimiser la stratégie d’agrégation. Ajuster le partitionnement si nécessaire. Réduire l’impact réseau. Livrable attendu : Proposer une optimisation permettant de limiter le shuffle et améliorer le temps d’exécution.

Contexte Un job analytique effectue trois jointures successives entre tables de grande taille, générant un temps d'exécution de plus de 45 minutes. Le Problème Chaque jointure déclenche un shuffle complet des données, aucune optimisation n'est appliquée. Contraintes Analyser le plan avec explain(True). Identifier les jointures éligibles au broadcast. Réordonner les jointures pour minimiser le volume shufflé. Livrable attendu Un script PySpark optimisé réduisant le nombre de shuffles via broadcast joins et réordonnancement des opérations.

Contexte : Une jointure entre deux DataFrames volumineux entraîne un ralentissement important. Le Problème : Certaines clés sont très fréquentes, provoquant un data skew et des tâches déséquilibrées. Extrait actuel : df1.join(df2, "user_id") Certaines valeurs de user_id dominent largement. Contraintes : Identifier le skew. Utiliser des techniques comme salting ou broadcast join si approprié. Minimiser le shuffle. Livrable attendu : Proposer une solution technique permettant de corriger le déséquilibre des partitions lors de la jointure.

Contexte Une équipe reporting souhaite générer un tableau croisé du chiffre d'affaires par produit et par trimestre à partir d'un DataFrame de ventes. Le Problème Le rapport est actuellement généré en pandas après un collect() sur un DataFrame de plusieurs millions de lignes. Contraintes Utiliser groupBy() et pivot() de PySpark directement sur le DataFrame distribué. Limiter les valeurs du pivot aux 4 trimestres connus pour éviter un pivot dynamique coûteux. Livrable attendu Un script PySpark générant un rapport pivot agrégé sans collect() préalable sur le DataFrame complet.

Contexte Un pipeline ETL charge quotidiennement des logs applicatifs et doit les transformer avant persistance. Le Problème Le pipeline actuel utilise un nombre de partitions par défaut inadapté, générant soit trop de petits fichiers soit des partitions trop volumineuses. Contraintes Utiliser repartition() ou coalesce() selon le contexte. Définir le nombre de partitions en fonction du volume estimé (128MB par partition). Justifier le choix entre repartition et coalesce. Livrable attendu Un pipeline ETL PySpark avec gestion explicite du nombre de partitions adaptée aux deux phases de transformation et d'écriture.

Contexte Une règle métier de catégorisation de risque client ne peut pas être exprimée avec les fonctions Spark natives. Le Problème L'équipe utilise une boucle Python sur le DataFrame pour appliquer la règle, ce qui contourne complètement le moteur distribué. Contraintes Créer un UDF avec udf() en spécifiant le type de retour. Enregistrer l'UDF pour usage SQL si nécessaire. Privilégier les fonctions natives Spark si possible, sinon justifier l'UDF. Livrable attendu Un script PySpark définissant et appliquant un UDF correctement typé pour la règle métier, avec note sur les implications performance.

Contexte Une équipe analytics souhaite calculer le rang des vendeurs par région ainsi que le chiffre d'affaires cumulé par mois. Le Problème Ces calculs nécessitent des fonctions analytiques que l'équipe tente d'implémenter avec des jointures multiples, générant un code illisible et peu performant. Contraintes Utiliser Window de pyspark.sql.window avec partitionBy() et orderBy(). Implémenter rank() pour le classement et sum() cumulatif avec rowsBetween(). Livrable attendu Un script PySpark utilisant les Window Functions pour calculer rang et cumul sans jointures redondantes.