Découvrez Apache Spark® 4.3 - Dès à présent disponible dans Databricks Runtime 19

Introduction

Apache Spark 4.3 intègre le traitement avec état (stateful) au Mode Temps Réel, atteignant une latence de bout en bout P99 inférieure à 150 ms sur trois benchmarks stateful représentatifs. Les agrégations en continu (streaming), la déduplication et le traitement stateful personnalisé peuvent désormais s'exécuter dans le mode d'exécution à faible latence de Structured Streaming, l'étendant ainsi aux applications qui ont besoin de mémoriser du contexte entre les événements.

L'impact va bien au-delà de pipelines plus rapides. Pour les charges de travail prises en charge, vous n'avez plus besoin d'un moteur de streaming séparé juste pour atteindre une latence de l'ordre de la milliseconde. Qu'il s'agisse de la détection des fraudes, de l'activité utilisateur mise à jour en continu ou de la surveillance d'appareils, les développeurs peuvent créer des applications stateful avec les API Structured Streaming familières dans le même moteur ouvert qu'ils utilisent pour le traitement par lots (batch), l'analytique et l'IA.

Le streaming en temps réel stateful est la vedette de Spark 4.3, mais les améliorations s'étendent à l'ensemble du moteur. La version apporte cinq avantages clés :

• Créer des applications en temps réel stateful : Utilisez l'agrégation en streaming, la déduplication et transformWithState en Mode Temps Réel pour maintenir le contexte à mesure que les événements arrivent.
• Préserver l'historique avec SCD Type 2 Auto CDC : Suivez l'évolution des enregistrements au fil du temps à l'aide de Python, SQL et Spark Connect, avec moins de logique de traitement des modifications personnalisée
• Itérer plus rapidement sur les petites données : Project Feather cible les surcharges d'exécution inutiles, tandis que la mise en cache native Arrow prend en charge une réutilisation des données plus efficace.
• Connecter plus d'applications à Spark : Les améliorations apportées à l'observabilité de Spark Connect, à l'intégration JDBC et à Python facilitent le développement et l'exploitation de l'exécution à distance.
• Exprimer des analyses plus riches directement en SQL : Les transformations VARIANT, JSON_TABLE, les jointures temporelles et un aperçu des horodatages à la nanoseconde étendent le traitement natif des données imbriquées et temporelles.

Ensemble, ces modifications aident les développeurs à utiliser un seul moteur ouvert pour explorer les données, maintenir un contexte changeant et alimenter des applications analytiques et d'IA.

Spark 4.3 marque également le début d'un rythme de publication trimestriel des fonctionnalités, passant d'une version tous les six mois à une version tous les trois mois. Ce cycle plus court permet aux utilisateurs de bénéficier plus rapidement des améliorations finalisées et offre à la communauté des opportunités plus fréquentes de donner son avis et de façonner la suite.

Streaming en Temps Réel : Applications State avec une Latence de l'Ordre de la Milliseconde

Du traitement d'événements au maintien du contexte

Un signal de fraude dépend non seulement de la dernière transaction, mais aussi de l'activité qui l'a précédée. La personnalisation doit refléter le comportement récent d'un utilisateur. La surveillance des appareils doit mettre à jour les compteurs et les agrégations à mesure que de nouvelles lectures arrivent. Ces applications nécessitent à la fois une faible latence et un état.

Spark 4.3 étend le Mode Temps Réel à ces schémoires de traitement stateful, en faisant un moteur de streaming complet. Il ajoute la prise en charge de l'agrégation en streaming, de dropDuplicates et de Scala/Java transformWithState (SPARK-58635, SPARK-57228). Les développeurs peuvent maintenir un état indexé (keyed state) et mettre à jour les résultats à l'aide du modèle de programmation existant de Structured Streaming, plutôt que d'adopter un autre moteur uniquement pour une plus faible latence.

Latence P99 inférieure à 150 ms sur trois benchmarks stateful

Dans les benchmarks rapportés dans SPARK-58884, les trois charges de travail stateful représentatives ont atteint une latence de bout en bout P99 inférieure à 150 millisecondes :

Ces résultats démontrent une exécution à faible latence pour la déduplication, l'agrégation par fenêtre et le traitement stateful personnalisé. Il s'agit de mesures des trois charges de travail testées, et non d'une garantie de latence pour chaque pipeline.

API familières, exécution à faible latence

Le Mode Temps Réel s'appuie sur Structured Streaming plutôt que d'introduire un modèle de programmation distinct. Les opérateurs stateful continuent d'utiliser les mécanismes de stockage d'état et de points de contrôle (checkpointing) existants de Spark pour la durabilité et la récupération.

Exemple : Maintenir un compteur continu pour chaque appareil. Étant donné un DataFrame de streaming adossé à Kafka events avec une colonne device_id :

Si des événements arrivent pour les appareils A, A et B, Spark maintient l'état et met à jour les comptes à A=2 et B=1. Lorsqu'un autre événement pour A arrive, son état est mis à jour à A=3.

Un moteur d'exécution conçu pour le flux de données en continu

La prise en charge du Mode Temps Réel stateful a nécessité un changement fondamental dans la façon dont Spark planifie et connecte les étapes (stages). Avec un shuffle bloquant traditionnel, une étape productrice doit se terminer avant que son étape aval ne puisse commencer. Ce modèle ne peut pas prendre en charge un flux de données continu entre des tâches de streaming de longue durée.

Spark 4.3 introduit la planification native des étapes simultanées (concurrent stage scheduling) et les shuffles en pipeline (pipelined shuffles), permettant aux étapes productrices et consommatrices de s'exécuter en même temps (SPARK-54699, SPARK-58076, SPARK-56664). Le Streaming Shuffle implémente cette capacité pour le Mode Temps Réel, en transférant les enregistrements de manière incrémentielle vers la tâche aval responsable de chaque clé au lieu d'attendre la fin de toute l'étape productrice.

Conceptuellement, l'exécution passe de :

Étape productrice → matérialiser le shuffle → étape consommatrice

à :

Étape productrice ⇄ streaming shuffle ⇄ étape consommatrice

Spark 4.3 ajoute PipelinedShuffleDependency, un nouveau type de dépendance de shuffle qui permet aux consommateurs de commencer à lire pendant que les producteurs s'exécutent encore (SPARK-58185). Le planificateur traite les étapes connectées par ces dépendances comme un groupe en pipeline, une unité de planification coordonnée. Ses étapes sont planifiées ensemble, et une défaillance dans une étape entraîne l'échec de tout le groupe (SPARK-58263, SPARK-58398). En bref, elles s'exécutent ensemble et échouent ensemble.

Étant donné que toutes les étapes d'un groupe en pipeline s'exécutent simultanément, Spark n'admet le groupe que lorsque suffisamment d'emplacements de tâches sont disponibles pour toutes ses tâches. Sinon, il échoue rapidement au lieu de se bloquer.

Spark achemine chaque shuffle vers un gestionnaire de shuffle (shuffle manager) en fonction de son type de dépendance :

• Un ShuffleDependency régulier utilise un gestionnaire de shuffle bloquant, qui matérialise toute la sortie du shuffle avant que les étapes aval ne la lisent. SortShuffleManager est l'implémentation intégrée.
• Un PipelinedShuffleDependency utilise un gestionnaire de mélange pipeline (pipelined shuffle manager), ce qui rend les données de sortie disponibles dès qu'elles sont produites. StreamingShuffleManager est l'implémentation intégrée (SPARK-57141).

Le type de dépendance définit le contrat d'exécution ; le gestionnaire de mélange fournit le mécanisme de transfert de données. L'ordonnanceur coordonne les étapes en pipeline sans avoir besoin de savoir quelle charge de travail les a créées.

Cette base va au-delà du mode temps réel (Real-Time Mode). L'ordonnancement de phases concurrentes, l'échange de données en pipeline et l'exécution coordonnée sont des fonctionnalités génériques du moteur Spark, et non une infrastructure de streaming ponctuelle. Dans Spark 4.3, elles permettent un traitement avec état à faible latence. À plus long terme, elles constituent une base pour d'autres charges de travail qui doivent s'affranchir du modèle d'exécution traditionnel étape par étape.

Pour en savoir plus sur la conception du mode temps réel et ses cas d'utilisation, consultez l'introduction au mode temps réel. Cet article décrit l'implémentation Databricks ; la portée des fonctionnalités de Spark 4.3 et les résultats de benchmark abordés ici sont ceux mentionnés ci-dessus.

Project Feather : Itérations plus rapides sur les petites données

Le développement assisté par LLM suit souvent une boucle de rétroaction courte : générer une transformation, l'exécuter sur un échantillon, inspecter les résultats et affiner le code. Chaque itération peut traiter seulement un petit fichier ou quelques lignes, mais l'ordonnancement des tâches et le mélange peuvent coûter plus cher que le calcul lui-même.

Project Feather cible cette surcharge d'exécution fixe, rendant Spark plus réactif pour explorer et valider du code SQL et DataFrame sur de petits jeux de données. Les développeurs peuvent utiliser les mêmes API lors du passage d'un échantillon à des charges de travail plus importantes, qu'ils écrivent le code eux-mêmes ou qu'ils travaillent avec un assistant IA.

Spark 4.3 introduit une exécution à tâche unique sans mélange, activable à la volée, pour les fragments de requêtes éligibles de petite taille et à partition unique (SPARK-57851). Lorsqu'un scan qualifié indique que sa sortie se trouve déjà dans une seule partition, les opérations aval prises en charge peuvent éviter un mélange inutile.

Exemple : Exécuter et inspecter une petite agrégation. Utilisez un petit jeu de données Parquet stocké sous forme de fichier unique, avec une colonne de catégorie :

Six cas :

category, Exchange, SPARK_LOCAL_CONNECT_REUSE=1, zip, counter_diff and jaro_winkler_similarity.

Le résultat est un décompte par catégorie. Pour une requête éligible, le plan physique peut omettre l'élément `Exchange` qui introduirait autrement un mélange. Cela prend en charge le même flux de travail d'exécution et d'inspection, que la transformation ait été écrite par un développeur ou suggérée par un assistant de code.

Avec les limites d'éligibilité par défaut, les scans de fichiers qualifiés lisent un petit fichier non vide qui tient dans une seule partition, tandis que les relations locales en mémoire qualifiées contiennent jusqu'à 1 000 lignes. Les types de requêtes pris en charge incluent les scans sélectionnés, les agrégations, les opérations distinctes, les fenêtres, les tris et les limites.

Feather ajoute également la mise en cache de jeux de données native Arrow facultative (SPARK-57268), stockant les lots mis en cache au format Arrow avec compression et statistiques de colonnes. Les entrées basées sur Arrow permettent d'éviter les conversions inutiles, prenant en charge les flux de travail qui explorent de manière répétée les mêmes données. L'optimisation à tâche unique et le cache Arrow sont tous deux optionnels. Pour les activer, définissez la configuration statique spark.sql.cache.serializer=org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer au démarrage d'une nouvelle application Spark. Un repartitionnement local intraprocessus plus large est prévu pour la prochaine version (SPARK-57399).

Ces améliorations ne sont qu'un début. La communauté Spark continuera d'investir dans les performances pour les petites données dans les versions futures, réduisant la surcharge d'exécution et raccourcissant les boucles de rétroaction pour l'exploration interactive et le développement assisté par LLM.

SCD Type 2 dans Auto CDC open source : Préserver l'historique au fil des modifications des données

Une table d'état actuel peut vous dire où habite un client aujourd'hui, mais pas à quelle région il appartenait lorsqu'une commande a été passée. Le reporting, l'audit et l'analyse historique nécessitent souvent les deux. Spark 4.3 ajoute la prise en charge complète du type Slowly Changing Dimension (SCD) 2 à Auto CDC (SPARK-56249), permettant aux pipelines de suivre l'historique de chaque enregistrement plutôt que d'écraser la dernière valeur à chaque modification. Cette fonctionnalité est disponible via Python, SQL et Spark Connect (SPARK-58321, SPARK-58247, SPARK-58261) et permet aux utilisateurs d'appliquer automatiquement les modifications au lieu d'effectuer des opérations de fusion manuelles.

Exemple : Suivre le statut et la région d'un client au fil du temps. Étant donné une source de pipeline customer_changes avec un identifiant client, un horodatage d'événement, un statut et une région :

Alternativement, cela peut être exprimé en SQL :

Dans une table de sortie SCD-2, chaque clé unique est représentée par une séquence d'enregistrements, chacun marquant le début et la fin de sa valabilité à l'aide de deux colonnes de métadonnées supplémentaires, __START_AT et __END_AT. Par exemple, lorsqu'un enregistrement de modification indique que le client avec l'identifiant 17 est passé de la région West à la région East, le pipeline conserve l'enregistrement avec la région précédente West et marque sa fin comme l'horodatage de l'événement de modification ; il insère un nouvel enregistrement avec la région East et l'horodatage de l'événement comme début. Une analyse peut ensuite utiliser la région qui s'appliquait au moment pertinent de la séquence d'événements.

Sous le capot, Auto CDC relie la définition de flux déclaratif à la dérivation de schéma cible et au traitement de flux SCD de type 2 dédié (SPARK-58319, SPARK-57395, SPARK-58321). Les clés identifient l'entité, le séquençage détermine l'ordre des modifications et les colonnes de suivi de l'historique spécifient quelles modifications d'attributs créent des versions historiques. Le moteur valide également les colonnes réservées, les types de séquençage et la configuration de suivi de l'historique (SPARK-57251, SPARK-58313, SPARK-58391), et réconcilie les micro-lots plus étroits avec un schéma cible évolué (SPARK-58418).

Auto CDC applique chaque micro-lot à la cible à l'aide de MERGE, par conséquent la table cible doit se trouver dans un catalogue dont le connecteur implémente les opérations au niveau des lignes de Data Source V2. Les formats de fichiers intégrés de Spark, y compris Parquet, ne sont pas éligibles. Par conséquent, un choix courant est Iceberg.

Spark Connect : Plus facile à développer, à observer et à exploiter

Les applications et les agents IA doivent invoquer Spark depuis leurs propres processus, gérer les sessions et relier une requête échouée au travail en cours sur le serveur. Spark 4.3 renforce Spark Connect pour ces flux de travail grâce à la visibilité des sessions et des opérations, à l'attribution des requêtes et à une prise en charge élargie des clients.

Spark Connect envoie des opérations DataFrame sous forme de plans logiques via gRPC, les exécute via le moteur Spark du serveur et transmet les résultats en continu par lots Arrow. Spark 4.3 ajoute des points de terminaison REST pour lister et inspecter les sessions et opérations Connect, ainsi que des identifiants d'opération pour tracer l'exécution entre les clients et le serveur (SPARK-57941, SPARK-58547). Pour le développement PySpark local sur les systèmes POSIX, l'option expérimentale `SPARK_LOCAL_CONNECT_REUSE=1` réutilise un serveur Connect persistant d'une exécution à l'autre, évitant ainsi la surcharge de démarrage répétée (SPARK-57787). Le wrapper de compatibilité SQLContext de PySpark fonctionne également avec Spark Connect (SPARK-57021).

Le pilote JDBC Spark Connect ajoute la prise en charge des résultats ARRAY, MAP et STRUCT ainsi qu'une gestion plus riche des jeux de résultats (SPARK-57380, SPARK-57274), des métadonnées de type SQL et une introspection de schéma plus compatible (SPARK-57627), ainsi qu'une gestion améliorée des exceptions (SPARK-57933, SPARK-58619). La couverture des API s'élargit avec DataFrame.zip, l'Auto CDC pour SCD de Type 2, ainsi que des types et littéraux d'horodatage en nanosecondes (SPARK-57247, SPARK-58247, SPARK-57160, SPARK-57161).

Par ailleurs, les travaux de la communauté incluent Spark Connect Gateway (SPARK-58455) pour le routage tenant compte des sessions et le fonctionnement multi-locataire, des clients Rust et Python basés sur Rust (SPARK-58960) pour une autre voie d'implémentation client, et une proposition de découverte des fonctionnalités (SPARK-56398) pour identifier les opérations de serveur prises en charge avant de soumettre une requête. Ces efforts ont des calendriers de développement distincts et ne sont pas groupés dans les fonctionnalités de la version 4.3.

PySpark : des API plus riches et une meilleure exécution Python

Les flux de travail Python créent souvent des colonnes dérivées en plusieurs étapes, combinent les résultats et déplacent les données entre pandas, Arrow et Spark. Spark 4.3 met une plus grande partie de ce travail à disposition via les API DataFrame tout en améliorant les chemins d'exécution sous-jacents. DataFrame.zip ajoute un moyen de combiner des projections de colonnes compatibles sans redescendre vers les API RDD, y compris lors de l'utilisation de Spark Connect (SPARK-55886, SPARK-57247).

Exemple : combinez des identifiants avec une valeur dérivée. Créez deux projections à partir de la même entrée et rassemblez leurs colonnes :

Le résultat associe id=1 à doubled=20 et id=2 à doubled=40. `zip` combine des transformations purement axées sur les projections du même plan sous-jacent ; le filtrage, le tri, les jointures ou les agrégations situés sous ces projections doivent être identiques des deux côtés pour préserver l'alignement des lignes.

Le travail d'exécution plus large cible le transfert de données, la création de lots et le calcul natif. Les lots d'entrée UDF basés sur Pickle peuvent être limités par octets plutôt que seulement par nombre de lignes, offrant un meilleur contrôle lorsque les lignes varient en taille (SPARK-57593). Les projections pandas-on-Spark par lots et les implémentations NumPy natives supplémentaires réduisent le travail évitable (SPARK-58230, SPARK-58234), tandis que la conversion de pandas vers Spark ajoute la prise en charge des colonnes PyArrow ChunkedArray (SPARK-46776).

La version ajoute également des liaisons Python et Scala pour les fonctions vectorielles existantes (SPARK-57284), la covariance groupée et les statistiques descriptives (SPARK-40336, SPARK-40337), les opérations de médiane mobile et d'extension ainsi que l'erreur type (SPARK-40341, SPARK-40340), et les diagnostics show_versions (SPARK-40373). Les variables locales de la trace de la pile (traceback) sont optionnelles car elles peuvent contenir des données sensibles (SPARK-57271).

Spark SQL : préparer un meilleur contexte pour l'analytique et l'IA

Les charges utiles d'événements arrivent sous forme de documents imbriqués, et des événements rapprochés peuvent nécessiter une plus grande précision que la microseconde. La préparation de ces entrées ne devrait pas nécessiter de couche de traitement personnalisée distincte pour chaque transformation. Spark 4.3 étend le SQL natif pour les données semi-structurées et l'analytique temporelle, incluant la manipulation de documents, la correspondance temporelle et un aperçu optionnel des horodatages en nanosecondes (SPARK-56822).

Exemple : distinguer deux événements au sein de la même microseconde. Activez l'aperçu (fonctionnalité non stable pour le moment, des modifications cassantes peuvent survenir) et comparez des horodatages séparés d'une nanoseconde (SPARK-56969, SPARK-56965, SPARK-57211) :

La comparaison donne true : le chiffre final est conservé plutôt que d'être ignoré. L'aperçu prend en charge TIMESTAMP_NTZ(p) et TIMESTAMP_LTZ(p) pour des précisions allant de 7 à 9.

Cela a nécessité bien plus que l'acceptation d'un nouveau nom de type SQL. La prise en charge des nanosecondes s'étend à l'analyse syntaxique (parsing), aux représentations de valeurs internes, aux conversions (casts), aux expressions, à Spark Connect, à la mise en cache et à l'intégration des sources de données (SPARK-56822). Les nouvelles fonctions timestamp_nanos et unix_nanos gèrent les conversions numériques (SPARK-57526, SPARK-57527). La prise en charge du stockage et des connecteurs reste spécifique à chaque chemin de données, l'aperçu doit donc être évalué de bout en bout.

Transformer des charges utiles JSON changeantes en données exploitables

Les événements d'application, les exportations de documents et les réponses de modèles arrivent souvent sous forme de JSON imbriqué. La fonction JSON_TABLE de Spark 4.3 (SPARK-58366) associe l'expansion de tableaux et l'extraction de champs typés, tandis que les transformations VARIANT permettent aux pipelines de mettre à jour ou de supprimer des champs sans UDF de traitement de documents personnalisés (SPARK-57804, SPARK-57417). Ensemble, ils simplifient la préparation du contenu pour les pipelines d'analytique, d'enrichissement et de récupération.

Exemple : Extraire les sections de documents pertinentes.

Cette opération produit une ligne typée qu'un pipeline peut enrichir avec des métadonnées source avant d'autres opérations en aval. JSON_TABLE résout ses chemins de colonnes via un parcours partagé de chaque élément de ligne, évitant une analyse séparée pour chaque colonne. Par ailleurs, Spark étend le pushdown d'extraction de champs VARIANT aux requêtes prises en charge qui effectuent des agrégations, des tris ou des jointures. (SPARK-58089).

Travaux SQL/JSON ultérieurs (SPARK-58686) prolongent cette direction avec JSON_VALUE, JSON_EXISTS et JSON_QUERY pour l'extraction de scalaires, les vérifications de présence et la récupération de fragments (SPARK-58685, SPARK-58687, SPARK-58723). Une gestion explicite des valeurs manquantes et des erreurs aidera les applications à traiter les schémas changeants de manière plus prévisible. Il s'agit de développements ultérieurs, et non de fonctionnalités supplémentaires de la version 4.3.

Faire correspondre les événements aux informations disponibles à ce moment-là

L'analyse historique et l'évaluation des modèles ont besoin du contexte disponible lorsqu'un événement s'est produit, et non des informations ajoutées plus tard. L'instruction SQL ASOF JOIN de Spark 4.3 exprime (fonctionnalité en version préliminaire, des modifications cassantes sont possibles) cette correspondance temporelle directement, remplaçant les jointures par plage écrites à la main suivies d'un classement (SPARK-58092).

Exemple : Reconstituer les caractéristiques disponibles pour une prédiction. Ici, available_at enregistre le moment où une version d'une caractéristique est devenue disponible (SPARK-58122, SPARK-58123) :

Une prédiction effectuée à 10h03 correspond à la version de la caractéristique disponible à 10h00, et non à 10h05. Un chemin d'exécution tri-fusion dédié fait de cette correspondance une opération du moteur plutôt qu'une logique de requête spécifique à l'application. Il en résulte un enrichissement ponctuel plus simple pour les données d'entraînement, les évaluations et l'analyse d'incidents, à condition que les horodatages de disponibilité représentent fidèlement ce qui était connu à ce moment-là.

Améliorations notables et remerciements

Spark 4.3 introduit également de nouvelles API et des améliorations de performance pour le traitement des données au quotidien. SQL intègre ANSI UNNEST pour développer des tableaux en lignes, ainsi que la fonction de fenêtre `counter_diff` pour convertir des compteurs cumulatifs en deltas et `jaro_winkler_similarity` pour les comparaisons de similarité de chaînes. Les sources de fichiers prises en charge ajoutent une option pour lire les archives tar, zip et 7z, éliminant ainsi une étape d'extraction séparée. Le moteur améliore l'exécution des agrégations et active par défaut la génération de code globale pour les opérations UNION éligibles. Une optimisation optionnelle répartit les clés de jointure nulles entre les partitions de mélange dans les jointures externes et anti éligibles. Le traitement des données imbriquées bénéficie d'un élagage accru des champs inutilisés dans les tableaux de structures et d'une génération de code pour davantage de fonctions d'ordre supérieur sur les tableaux. Les lectures Parquet vectorisées améliorent la conversion de types et le décodage de dictionnaires.

Spark 4.3 reflète la force de la communauté Apache Spark, avec plus de 200 contributeurs ayant aidé à traiter plus de 1 100 tickets Jira. Nous remercions tous ceux qui ont contribué par du code, des relectures, des tests, de la documentation et des retours pour rendre cette version possible.

Commencer avec Spark 4.3

Visitez la page de téléchargement d'Apache Spark pour obtenir les téléchargements de versions et la documentation, et consultez les guides de migration avant la mise à niveau. Pour connaître la disponibilité sur Databricks, consultez les notes de version de Databricks Runtime.

(Cet article de blog a été traduit à l'aide d'outils basés sur l'intelligence artificielle) Article original