Accueil›Blog›Test technique Spark avance : Structured Streaming, Delta Lake, optimisation

Guide recrutement data

Test technique Spark avance : Structured Streaming, Delta Lake, optimisation

Au-dela des DataFrames de base, un Data Engineer Spark Senior maitrise le streaming, Delta Lake et l optimisation fine des jobs. C est ce qu on evalue pour les profils qui font tourner Spark en production.

Data Builder·Juin 2025·7 min de lecture·Data Engineer

Sommaire

Structured Streaming
Delta Lake
Optimisation avancee AQE
Spark on Kubernetes
Memoire et GC tuning
Unity Catalog et governance
Grille

Spark batch sur des fichiers CSV, tout le monde sait faire. Ce qu on evalue en entretien Senior, c est le streaming, la gestion des transactions ACID avec Delta Lake et la capacite a diagnostiquer des jobs lents.

1Structured Streaming : traitement en continu

Question discriminante

Quelle est la difference entre le micro-batch et le continuous processing dans Spark Structured Streaming ?

from pyspark.sql import SparkSession
from pyspark.sql.functions import window, count

spark = SparkSession.builder.appName('streaming').getOrCreate()

# Lire depuis Kafka en streaming
df = spark.readStream \
    .format('kafka') \
    .option('kafka.bootstrap.servers', 'kafka:9092') \
    .option('subscribe', 'events') \
    .load()

# Aggregation avec window temporelle
result = df \
    .selectExpr('CAST(value AS STRING) as event') \
    .groupBy(window('timestamp', '5 minutes'), 'event') \
    .agg(count('*').alias('cnt'))

# Ecriture en Delta Lake
query = result.writeStream \
    .format('delta') \
    .outputMode('complete') \
    .option('checkpointLocation', '/checkpoints/events') \
    .start('/data/events_aggregated')

query.awaitTermination()

Micro-batch — par defaut, Spark traite de petits lots a intervalle fixe. Latence de quelques secondes
Continuous processing — mode experimental, latence sub-milliseconde, operations limitees
Checkpoint — stocke l etat du streaming pour la reprise apres panne. Obligatoire en production
Watermark — gerer les evenements en retard (late data) dans les aggregations temporelles

2Delta Lake : ACID pour le data lake

Question discriminante

Quels problemes Delta Lake resout-il par rapport a Parquet classique ?

from delta.tables import DeltaTable

# MERGE : upsert ACID
delta_table = DeltaTable.forPath(spark, '/data/customers')

delta_table.alias('target').merge(
    updates_df.alias('source'),
    'target.customer_id = source.customer_id'
).whenMatchedUpdate(set={
    'name': 'source.name',
    'email': 'source.email',
    'updated_at': 'source.updated_at'
}).whenNotMatchedInsert(values={
    'customer_id': 'source.customer_id',
    'name': 'source.name',
    'email': 'source.email',
    'updated_at': 'source.updated_at'
}).execute()

# Time Travel : requeter les donnees passees
df_yesterday = spark.read.format('delta') \
    .option('timestampAsOf', '2025-01-01') \
    .load('/data/customers')

# OPTIMIZE + ZORDER : ameliorer les perfs de lecture
delta_table.optimize().executeZOrderBy('region', 'date')

ACID — transactions atomiques. Plus de fichiers corrompus si le job crashe en cours
Time Travel — requeter les donnees a n importe quelle version ou timestamp precedent
Schema Evolution — ajouter des colonnes sans casser les requetes existantes
MERGE — upsert en une seule operation. Essentiel pour les CDC (Change Data Capture)

3Adaptive Query Execution (AQE)

Question discriminante

Qu est-ce que l AQE dans Spark 3+ ? Comment resout-il le probleme du data skew ?

# Activer AQE (actif par defaut dans Spark 3.2+)
spark.conf.set('spark.sql.adaptive.enabled', 'true')
spark.conf.set('spark.sql.adaptive.coalescePartitions.enabled', 'true')
spark.conf.set('spark.sql.adaptive.skewJoin.enabled', 'true')

# Pour le skew : diviser automatiquement les partitions desequilibrees
spark.conf.set('spark.sql.adaptive.skewJoin.skewedPartitionFactor', '5')
spark.conf.set('spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes', '256m')

AQE — Spark optimise le plan d execution a la volee, en tenant compte des statistiques reelles des donnees intermediaires
Skew join handling — detecte et divise automatiquement les partitions tres lourdes (data skew)
Dynamic coalescing — reduit le nombre de partitions shuffle si les donnees sont petites
Broadcast join — convertit automatiquement un sort-merge join en broadcast join si une table est petite

4Spark on Kubernetes

spark-submit --master k8s:// — le driver et les executors sont des Pods Kubernetes
Avantages — isolation par job, scaling elastique, meilleure utilisation des ressources vs YARN
Spot instances — reduire les couts de 60-80% sur les jobs batch en utilisant des noeuds preemptibles
Volcano scheduler — scheduler Kubernetes optimise pour les workloads ML/data (gang scheduling)

5Memory et GC tuning

# Configuration memoire Spark
spark.conf.set('spark.executor.memory', '8g')
spark.conf.set('spark.executor.memoryOverhead', '2g')  # memoire off-heap
spark.conf.set('spark.sql.shuffle.partitions', '200')  # ajuster selon le volume

# Preferer G1GC pour les gros heap
spark.conf.set('spark.executor.extraJavaOptions', '-XX:+UseG1GC')

# Persistance strategique
df_heavy.persist(StorageLevel.MEMORY_AND_DISK)  # si reutilise plusieurs fois
# Nettoyer apres usage
df_heavy.unpersist()

6Unity Catalog : gouvernance Databricks

Unity Catalog — couche de gouvernance centralisee pour tous les assets Databricks (tables Delta, fichiers, modeles ML, notebooks)
3 niveaux — Catalog > Schema > Table. Remplace le Hive metastore
Column-level security — masquer des colonnes selon le role de l utilisateur
Data lineage automatique — lineage jusqu a la colonne, integre avec dbt et Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, broadcast

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

# Broadcast join : diffuser la petite table sur tous les workers
result = large_df.join(broadcast(small_df), 'category_id')

# Repartition strategique avant un groupBy lourd
df_repartitioned = df.repartition(200, col('customer_id'))
agg_result = df_repartitioned.groupBy('customer_id').agg({'amount': 'sum'})

# Persist pour reutiliser un DataFrame calcule plusieurs fois
df_filtered = df.filter(col('status') == 'completed').persist()
count = df_filtered.count()
agg = df_filtered.groupBy('region').sum('amount')

# Lire le plan d execution
result.explain(extended=True)  # voir les stages, shuffles, pushdowns

AQE (Adaptive Query Execution) - Spark 3.x optimise le plan d execution au runtime. Coalesce les partitions vides, detecte le data skew, choisit broadcast join automatiquement
Shuffle partitions - spark.sql.shuffle.partitions=200 par defaut. Regler selon la volumetrie : 1 partition ~128MB de donnees shufflees est un bon point de depart
Data skew - une partition 100x plus grosse que les autres = un executor qui bloque tous les autres. Remedies : salting (ajouter du bruit a la cle de join), AQE skewJoin, ou filter + union
Catalyst optimizer - Spark convertit les DataFrames en plan logique puis physique optimise. Privilegier les DataFrames sur les RDDs pour profiter de Catalyst
Tungsten - moteur d execution memoire de Spark. Serialisation binaire, generation de bytecode. Utilise automatiquement avec les DataFrames

7Grille par niveau

Niveau	Maitrise	Signal GO	NO-GO
Confirme	DataFrame API, Delta Lake basique, partitionnement	A utilise Delta MERGE, sait ce qu est Time Travel, configure les partitions shuffle	N a jamais utilise Delta Lake, ne sait pas ce qu est AQE
Senior	Structured Streaming, AQE, Spark on K8s, memory tuning	A deploye un job Spark Streaming, a debug un skew join, a optimise la memoire	N a jamais fait de streaming, ne sait pas diagnostiquer un OOM Spark
Lead	Architecture lakehouse, Unity Catalog, standards d equipe	A concu une architecture lakehouse, defini les standards de gouvernance Databricks	Ne peut pas expliquer la difference entre Data Lake et Lakehouse

Spark batch on CSV files, everyone knows how to do it. What we evaluate in Senior interviews is streaming, ACID transaction management with Delta Lake and the ability to diagnose slow jobs.

1Structured Streaming: continuous processing

Discriminating question

What is the difference between micro-batch and continuous processing in Spark Structured Streaming?

from pyspark.sql import SparkSession
from pyspark.sql.functions import window, count

spark = SparkSession.builder.appName('streaming').getOrCreate()

# Read from Kafka in streaming
df = spark.readStream \
    .format('kafka') \
    .option('kafka.bootstrap.servers', 'kafka:9092') \
    .option('subscribe', 'events') \
    .load()

# Aggregation with time window
result = df \
    .selectExpr('CAST(value AS STRING) as event') \
    .groupBy(window('timestamp', '5 minutes'), 'event') \
    .agg(count('*').alias('cnt'))

# Write to Delta Lake
query = result.writeStream \
    .format('delta') \
    .outputMode('complete') \
    .option('checkpointLocation', '/checkpoints/events') \
    .start('/data/events_aggregated')

query.awaitTermination()

Micro-batch — by default, Spark processes small batches at fixed intervals. Latency of a few seconds
Continuous processing — experimental mode, sub-millisecond latency, limited operations
Checkpoint — stores the streaming state for recovery after failure. Mandatory in production
Watermark — handle late data events in time-based aggregations

2Delta Lake: ACID for the data lake

Discriminating question

What problems does Delta Lake solve compared to classic Parquet?

from delta.tables import DeltaTable

# MERGE: ACID upsert
delta_table = DeltaTable.forPath(spark, '/data/customers')

delta_table.alias('target').merge(
    updates_df.alias('source'),
    'target.customer_id = source.customer_id'
).whenMatchedUpdate(set={
    'name': 'source.name',
    'email': 'source.email',
    'updated_at': 'source.updated_at'
}).whenNotMatchedInsert(values={
    'customer_id': 'source.customer_id',
    'name': 'source.name',
    'email': 'source.email',
    'updated_at': 'source.updated_at'
}).execute()

# Time Travel: query past data
df_yesterday = spark.read.format('delta') \
    .option('timestampAsOf', '2025-01-01') \
    .load('/data/customers')

# OPTIMIZE + ZORDER: improve read performance
delta_table.optimize().executeZOrderBy('region', 'date')

ACID — atomic transactions. No more corrupted files if the job crashes midway
Time Travel — query data at any previous version or timestamp
Schema Evolution — add columns without breaking existing queries
MERGE — upsert in a single operation. Essential for CDC (Change Data Capture)

3Adaptive Query Execution (AQE)

Discriminating question

What is AQE in Spark 3+? How does it solve the data skew problem?

# Enable AQE (active by default in Spark 3.2+)
spark.conf.set('spark.sql.adaptive.enabled', 'true')
spark.conf.set('spark.sql.adaptive.coalescePartitions.enabled', 'true')
spark.conf.set('spark.sql.adaptive.skewJoin.enabled', 'true')

# For skew: automatically split unbalanced partitions
spark.conf.set('spark.sql.adaptive.skewJoin.skewedPartitionFactor', '5')
spark.conf.set('spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes', '256m')

AQE — Spark optimizes the execution plan on the fly, taking into account real statistics from intermediate data
Skew join handling — automatically detects and splits very large partitions (data skew)
Dynamic coalescing — reduces the number of shuffle partitions if data is small
Broadcast join — automatically converts a sort-merge join into a broadcast join if one table is small

4Spark on Kubernetes

spark-submit --master k8s:// — the driver and executors are Kubernetes Pods
Advantages — per-job isolation, elastic scaling, better resource utilization vs YARN
Spot instances — reduce costs by 60-80% on batch jobs using preemptible nodes
Volcano scheduler — Kubernetes scheduler optimized for ML/data workloads (gang scheduling)

5Memory and GC tuning

# Spark memory configuration
spark.conf.set('spark.executor.memory', '8g')
spark.conf.set('spark.executor.memoryOverhead', '2g')  # off-heap memory
spark.conf.set('spark.sql.shuffle.partitions', '200')  # adjust based on volume

# Prefer G1GC for large heaps
spark.conf.set('spark.executor.extraJavaOptions', '-XX:+UseG1GC')

# Strategic persistence
df_heavy.persist(StorageLevel.MEMORY_AND_DISK)  # if reused multiple times
# Clean up after use
df_heavy.unpersist()

6Unity Catalog: Databricks governance

Unity Catalog — centralized governance layer for all Databricks assets (Delta tables, files, ML models, notebooks)
3 levels — Catalog > Schema > Table. Replaces the Hive metastore
Column-level security — mask columns based on the user's role
Automatic data lineage — lineage down to the column level, integrated with dbt and Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, broadcast

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

# Broadcast join: broadcast the small table to all workers
result = large_df.join(broadcast(small_df), 'category_id')

# Strategic repartition before a heavy groupBy
df_repartitioned = df.repartition(200, col('customer_id'))
agg_result = df_repartitioned.groupBy('customer_id').agg({'amount': 'sum'})

# Persist to reuse a DataFrame computed multiple times
df_filtered = df.filter(col('status') == 'completed').persist()
count = df_filtered.count()
agg = df_filtered.groupBy('region').sum('amount')

# Read the execution plan
result.explain(extended=True)  # view stages, shuffles, pushdowns

AQE (Adaptive Query Execution) - Spark 3.x optimizes the execution plan at runtime. Coalesces empty partitions, detects data skew, automatically chooses broadcast join
Shuffle partitions - spark.sql.shuffle.partitions=200 by default. Tune based on data volume: 1 partition ~128MB of shuffled data is a good starting point
Data skew - one partition 100x larger than the others = one executor blocking all the others. Remedies: salting (adding noise to the join key), AQE skewJoin, or filter + union
Catalyst optimizer - Spark converts DataFrames into a logical then optimized physical plan. Prefer DataFrames over RDDs to benefit from Catalyst
Tungsten - Spark's in-memory execution engine. Binary serialization, bytecode generation. Used automatically with DataFrames

7Level grid

Level	Mastery	GO signal	NO-GO
Mid-level	DataFrame API, basic Delta Lake, partitioning	Has used Delta MERGE, knows what Time Travel is, configures shuffle partitions	Has never used Delta Lake, does not know what AQE is
Senior	Structured Streaming, AQE, Spark on K8s, memory tuning	Has deployed a Spark Streaming job, has debugged a skew join, has optimized memory	Has never done streaming, cannot diagnose a Spark OOM
Lead	Lakehouse architecture, Unity Catalog, team standards	Has designed a lakehouse architecture, defined Databricks governance standards	Cannot explain the difference between Data Lake and Lakehouse

Vous recrutez un Data Engineer Spark ?

Premier entretien gratuit. Rapport GO/NO-GO sous 48h.

Tester gratuitement Reserver un appel