Apache Spark 4 en 2026 : Nouvelles Fonctionnalités, Structured Streaming et Questions d'Entretien

Apache Spark 4 représente la refonte la plus importante du moteur Spark depuis la version 3.0. Publié mi-2025 avec une itération rapide jusqu'à la version 4.1 et la prochaine 4.2, la lignée 4.x redéfinit la manière dont les ingénieurs data construisent des pipelines, exécutent des charges de travail streaming et manipulent du SQL à grande échelle. Ce guide détaille les changements critiques, les démontre avec du code PySpark et Scala, et couvre les questions d'entretien les plus posées par les équipes de recrutement.

Le mode ANSI SQL activé par défaut dans Spark 4

Spark 4.0 bascule spark.sql.ansi.enabled sur true. Chaque dépassement arithmétique, cast invalide et division par zéro qui renvoyait précédemment NULL ou tronquait silencieusement lève désormais une exception. L'objectif : une conformité stricte au standard SQL correspondant au comportement de PostgreSQL et Oracle.

L'impact pratique touche principalement les pipelines ETL. Les requêtes qui s'appuyaient sur la coercition implicite ou la propagation silencieuse de NULL échoueront à l'exécution jusqu'à ce qu'elles soient explicitement gérées.

# ansi_mode_migration.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, try_cast

spark = SparkSession.builder.appName("ANSIMigration").getOrCreate()

# Ceci lève maintenant une ArithmeticException en mode ANSI Spark 4
# SELECT CAST('abc' AS INT)  -- erreur à l'exécution

# Migration sécurisée : utiliser TRY_CAST pour préserver le comportement NULL
df = spark.sql("""
    SELECT
        TRY_CAST(raw_value AS INT) AS parsed_value,
        TRY_CAST(amount AS DECIMAL(10,2)) AS safe_amount
    FROM raw_events
    WHERE TRY_CAST(raw_value AS INT) IS NOT NULL
""")

# Alternativement, envelopper l'arithmétique dans try_divide / try_add
result = df.withColumn(
    "ratio",
    col("safe_amount") / col("parsed_value")  # lève une exception sur zéro
)

Checklist de migration : auditer chaque appel CAST, remplacer par TRY_CAST là où un comportement NULL-safe est attendu, et ajouter des vérifications explicites de dépassement sur les colonnes numériques. Le guide de migration Spark SQL documente chaque changement de comportement.

Le type VARIANT pour les données semi-structurées

Spark 4.0 ajoute un type de colonne natif VARIANT conçu pour les données semi-structurées de type JSON. Contrairement au stockage de chaînes JSON brutes avec parsing au moment de la requête, VARIANT utilise un format binaire optimisé avec déchiquetage de colonnes qui offre des lectures jusqu'à 8 fois plus rapides lors de l'accès aux champs imbriqués.

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder.appName("VariantDemo").getOrCreate()
import spark.implicits._

// Créer une table avec une colonne VARIANT
spark.sql("""
    CREATE TABLE events (
        event_id BIGINT,
        timestamp TIMESTAMP,
        payload VARIANT
    ) USING PARQUET
""")

// Insérer des données JSON — automatiquement converties en VARIANT
spark.sql("""
    INSERT INTO events VALUES
    (1, current_timestamp(),
     PARSE_JSON('{"user_id": 42, "action": "click", "metadata": {"page": "/home", "duration_ms": 320}}')),
    (2, current_timestamp(),
     PARSE_JSON('{"user_id": 87, "action": "purchase", "metadata": {"item_id": "SKU-9001", "amount": 49.99}}'))
""")

// Interroger les champs imbriqués avec la notation point — utilise le déchiquetage pour un accès rapide
val clicks = spark.sql("""
    SELECT
        event_id,
        payload:user_id::INT AS user_id,
        payload:action::STRING AS action,
        payload:metadata.page::STRING AS page
    FROM events
    WHERE payload:action::STRING = 'click'
""")
clicks.show()

VARIANT élimine le besoin de contournements schema-on-read. Dans les architectures data lake où les schémas d'événements évoluent fréquemment, une seule colonne VARIANT absorbe les changements de schéma sans migrations DDL. Spark 4.1 a étendu VARIANT avec du déchiquetage au niveau Parquet, réduisant encore davantage les IO pour l'accès sélectif aux champs.

Le streaming en mode Real-Time dans Spark 4.1

Structured Streaming dans Spark 4.1 introduit le mode Real-Time (RTM), le premier support officiel pour une latence inférieure à la seconde dans Spark. Pour les charges de travail stateless, la latence descend à quelques millisecondes, en concurrence directe avec Apache Flink sur son propre terrain.

Trois changements architecturaux rendent cela possible : le flux de données continu (pas de frontières micro-batch), la planification de pipeline (chevauchement des étapes lecture/calcul/écriture), et le shuffle streaming (état stocké en mémoire plutôt que checkpointé entre les batchs).

# real_time_streaming.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, window
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

spark = SparkSession.builder \
    .appName("RTMDemo") \
    .config("spark.sql.streaming.mode", "realtime") \
    .getOrCreate()

schema = StructType([
    StructField("sensor_id", StringType()),
    StructField("temperature", DoubleType()),
    StructField("event_time", TimestampType())
])

# Lire depuis Kafka avec le mode Real-Time activé
raw_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("subscribe", "sensor-readings") \
    .option("startingOffsets", "latest") \
    .load()

parsed = raw_stream \
    .select(from_json(col("value").cast("string"), schema).alias("data")) \
    .select("data.*")

# Écrire vers un sink Kafka — end-to-end en moins d'une seconde
query = parsed \
    .filter(col("temperature") > 80.0) \
    .writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("topic", "alerts") \
    .option("checkpointLocation", "/tmp/checkpoints/alerts") \
    .start()

query.awaitTermination()

Activer RTM nécessite un seul changement de configuration : spark.sql.streaming.mode = realtime. Aucune réécriture de code n'est nécessaire. Spark 4.1 commence avec des requêtes stateless, single-stage en Scala et PySpark utilisant des sources Kafka et des sinks Kafka/Foreach. Le support RTM stateful est attendu dans Spark 4.2.

Spark Connect et le client PySpark léger

Spark 4.0 livre un package pyspark-client autonome pesant seulement 1,5 Mo, comparé à la distribution PySpark complète de plus de 300 Mo. Ce client léger communique avec un cluster Spark distant via gRPC à travers Spark Connect, permettant aux environnements notebook et aux pipelines CI d'exécuter des jobs Spark sans embarquer le runtime complet.

# spark_connect_client.py
from pyspark.sql import SparkSession

# Se connecter à un cluster Spark distant — pas besoin de JVM locale
spark = SparkSession.builder \
    .remote("sc://spark-cluster.internal:15002") \
    .getOrCreate()

# API DataFrame complète disponible via le client léger
df = spark.read.parquet("s3a://data-lake/events/2026/")

aggregated = df.groupBy("country", "event_type") \
    .count() \
    .orderBy("count", ascending=False)

aggregated.show(20)

Spark Connect supporte également les clients Java et Swift. La configuration spark.api.mode permet de basculer entre le mode embedded classique et le mode Connect sans changement de code.

L'API transformWithState pour une logique streaming complexe

Spark 4.0 remplace le limité mapGroupsWithState par transformWithState, une API flexible supportant les types d'état composites, l'expiration basée sur TTL, et les timers événementiels.

import org.apache.spark.sql.streaming._
import org.apache.spark.sql.streaming.StatefulProcessor
import java.time.Duration

case class UserEvent(userId: String, action: String, timestamp: Long)
case class SessionSummary(userId: String, actionCount: Int, durationMs: Long)

class SessionProcessor extends StatefulProcessor[String, UserEvent, SessionSummary] {

  @transient private var sessionStart: ValueState[Long] = _
  @transient private var actionCount: ValueState[Int] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    sessionStart = getHandle.getValueState[Long]("start", TTLConfig(Duration.ofMinutes(30)))
    actionCount = getHandle.getValueState[Int]("count", TTLConfig(Duration.ofMinutes(30)))
  }

  override def handleInputRows(
      key: String,
      rows: Iterator[UserEvent],
      timerValues: TimerValues
  ): Iterator[SessionSummary] = {
    rows.flatMap { event =>
      if (!sessionStart.exists()) {
        sessionStart.update(event.timestamp)
        actionCount.update(1)
        Iterator.empty
      } else {
        val count = actionCount.get() + 1
        actionCount.update(count)
        if (count % 10 == 0) {
          val duration = event.timestamp - sessionStart.get()
          Iterator(SessionSummary(key, count, duration))
        } else Iterator.empty
      }
    }
  }
}

Les Spark Declarative Pipelines (SDP) dans Spark 4.1

Spark 4.1 ajoute les Spark Declarative Pipelines, un framework où les ingénieurs définissent les datasets et transformations tandis que Spark gère l'ordonnancement de l'exécution, le parallélisme, les checkpoints et les retries.

# declarative_pipeline.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as _sum, current_date

spark = SparkSession.builder.appName("SDP-Demo").getOrCreate()

@spark.declarative.table("clean_orders")
def clean_orders():
    return spark.read.table("raw_orders") \
        .filter(col("status") != "cancelled") \
        .filter(col("amount") > 0)

@spark.declarative.table("daily_revenue")
def daily_revenue():
    return spark.read.table("clean_orders") \
        .groupBy("order_date", "region") \
        .agg(_sum("amount").alias("total_revenue"))

@spark.declarative.table("revenue_summary")
def revenue_summary():
    return spark.read.table("daily_revenue") \
        .groupBy("region") \
        .agg(_sum("total_revenue").alias("grand_total")) \
        .orderBy("grand_total", ascending=False)

spark.declarative.run()

Questions d'entretien data engineering sur Spark 4

Les équipes de recrutement évaluent désormais la maîtrise de Spark 4 à travers des questions ciblées sur les changements architecturaux et les nouvelles capacités.

Q1 : Comment le mode ANSI SQL de Spark 4 affecte-t-il les pipelines ETL existants et quelle est la stratégie de migration ?

Le mode ANSI fait passer Spark d'un comportement permissif (NULL sur erreur) à un comportement strict (exception sur erreur). Les pipelines qui dépendent de conversions implicites ou de dépassements silencieux échoueront. La stratégie de migration implique : auditer tous les CAST et les remplacer par TRY_CAST là où NULL est acceptable, ajouter des validations explicites sur les colonnes numériques, et tester avec spark.sql.ansi.enabled=true en staging avant la production. Les fonctions try_* (try_divide, try_add) permettent une transition progressive.

Q2 : Expliquez le type de données VARIANT et ses avantages par rapport au stockage JSON en chaîne de caractères.

VARIANT est un type de colonne natif pour les données semi-structurées utilisant un format binaire optimisé avec déchiquetage de colonnes. Contrairement aux chaînes JSON qui nécessitent un parsing à chaque requête, VARIANT analyse une fois à l'ingestion et stocke les champs fréquemment accédés dans des colonnes séparées au niveau Parquet. Cela réduit les IO jusqu'à 8x pour l'accès sélectif aux champs. VARIANT absorbe également les changements de schéma sans migrations DDL, ce qui est idéal pour les architectures event-driven où les schémas évoluent.

Q3 : Quelles sont les limites du mode Real-Time dans Spark 4.1 et quand devrait-on utiliser le micro-batching traditionnel ?

Real-Time Mode dans Spark 4.1 est limité aux requêtes stateless, single-stage avec des sources Kafka et des sinks Kafka/Foreach. Les opérations stateful (windowing, aggregations avec état) ne sont pas encore supportées. Le micro-batching reste préférable pour : les workloads stateful complexes, les scénarios nécessitant exactly-once semantics avec state stores, et les pipelines où la latence sub-seconde n'est pas critique. RTM brille pour les alertes en temps réel simples et le filtrage à haute fréquence.

Q4 : Comment fonctionne Spark Connect et quels sont ses cas d'usage principaux ?

Spark Connect sépare le client de l'exécution en utilisant gRPC pour la communication. Le client pyspark-client (1,5 Mo) envoie des plans logiques à un cluster Spark distant qui effectue l'exécution réelle. Les cas d'usage incluent : des notebooks légers sans dépendances JVM locales, des pipelines CI/CD où l'empreinte Docker doit être minimale, et des clients multi-langage (Java, Swift) se connectant au même cluster. Connect permet également le multiplexing de sessions et l'isolation des requêtes.

Q5 : Décrivez l'architecture des Spark Declarative Pipelines et comment elles gèrent les dépendances.

Spark Declarative Pipelines (SDP) utilisent des décorateurs Python ou des annotations Scala pour marquer les fonctions comme des datasets. Le framework Spark construit un DAG de dépendances en analysant les appels spark.read.table() au sein de chaque fonction. À l'exécution, Spark ordonnance automatiquement les tâches en ordre topologique, parallélise les branches indépendantes, et gère les checkpoints. SDP élimine la logique d'orchestration manuelle et permet de rejouer sélectivement les datasets en aval lorsque les sources changent.

Construire un pipeline ETL complet avec Spark 4

Un pipeline de production combinant les fonctionnalités Spark 4 démontre l'intégration du mode ANSI, du type VARIANT et du streaming en temps réel.

# spark4_etl_pipeline.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, current_timestamp, parse_json

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

bronze = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("subscribe", "user-events") \
    .load() \
    .select(
        col("key").cast("string").alias("event_key"),
        parse_json(col("value").cast("string")).alias("payload"),
        col("timestamp").alias("kafka_ts")
    )

silver = bronze.select(
    col("event_key"),
    col("payload:user_id").cast("int").alias("user_id"),
    col("payload:action").cast("string").alias("action"),
    col("payload:metadata.session_id").cast("string").alias("session_id"),
    col("kafka_ts"),
    current_timestamp().alias("processed_at")
).filter(
    col("user_id").isNotNull()
)

query = silver.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "s3a://checkpoints/user-events/") \
    .option("mergeSchema", "true") \
    .toTable("gold.user_events")

query.awaitTermination()

Conclusion

Apache Spark 4 marque une évolution majeure de l'écosystème du traitement de données distribuées. Les changements clés que les ingénieurs data doivent maîtriser incluent :

• Mode ANSI SQL : conformité stricte nécessitant une migration explicite des pipelines existants avec TRY_CAST et validation des dépassements
• Type VARIANT : accélération jusqu'à 8x pour les données semi-structurées grâce au déchiquetage de colonnes au niveau Parquet
• Real-Time Mode : latence sub-seconde pour les workloads stateless, comblant l'écart avec les moteurs de streaming dédiés
• Spark Connect : clients légers permettant l'exécution distante sans dépendances JVM complètes
• transformWithState : logique streaming stateful flexible avec support TTL et timers événementiels
• Declarative Pipelines : orchestration automatique des DAG de datasets éliminant la gestion manuelle des dépendances

La roadmap Spark 4.2 promet d'étendre Real-Time Mode aux opérations stateful et d'améliorer les performances VARIANT avec des codecs de compression adaptatifs. Les organisations adoptant Spark 4 bénéficient d'une conformité SQL améliorée, d'une ingestion de données semi-structurées plus rapide, et d'options streaming répondant à des exigences de latence diverses.