Moteurs de requête : Trino, Spark, Dremio, Flink, DuckDB.
Cette compétence couvre les trois formats leaders (Delta, Iceberg, Hudi), les catalogs, les opérations de maintenance (compaction, vacuum, expiration de snapshots), les optimisations de performance (Z-order, partitionnement évolutif), et l'intégration avec les moteurs de requête modernes.
Quand l'utiliser
Activez cette compétence lorsque l'utilisateur :
Conçoit une architecture de données moderne (lakehouse).
Doit choisir entre Delta Lake, Iceberg ou Hudi.
Veut migrer d'un data lake traditionnel (fichiers HDFS bruts) vers un lakehouse ACID.
Pose des questions sur Time Travel, schema evolution, compaction, vacuum.
# Création d'une table Delta
df.write.format("delta") \
.mode("overwrite") \
.partitionBy("date") \
.option("delta.tuneFileSizesForRewrites", "true") \
.save("/data/lake/bronze/sensors/")
# Append avec mergeSchema (ajout automatique de colonnes)
df_new.write.format("delta") \
.mode("append") \
.option("mergeSchema", "true") \
.save("/data/lake/bronze/sensors/")
2.3 MERGE (Upsert)
from delta.tables import DeltaTable
delta_table = DeltaTable.forPath(spark, "/data/lake/silver/sensors")
# Upsert : mettre à jour les lignes existantes, insérer les nouvelles
delta_table.alias("target") \
.merge(
df_updates.alias("source"),
"target.machine_id = source.machine_id AND target.timestamp = source.timestamp"
) \
.whenMatchedUpdate(set={
"temperature": "source.temperature",
"pression": "source.pression",
"updated_at": "current_timestamp()"
}) \
.whenNotMatchedInsert(values={
"machine_id": "source.machine_id",
"temperature": "source.temperature",
"pression": "source.pression",
"timestamp": "source.timestamp",
"created_at": "current_timestamp()"
}) \
.execute()
2.4 Time Travel
# Lire une version antérieure
df_v1 = spark.read.format("delta") \
.option("versionAsOf", "1") \
.load("/data/lake/silver/sensors/")
# Lire à une date
df_ts = spark.read.format("delta") \
.option("timestampAsOf", "2026-07-20") \
.load("/data/lake/silver/sensors/")
# Lister l'historiquefrom delta.tables import DeltaTable
history = DeltaTable.forPath(spark, "/data/lake/silver/sensors/").history()
history.select("version", "timestamp", "operation", "operationParameters").show()
2.5 Optimisation : Z-Order et Compaction
from delta.tables import DeltaTable
delta_table = DeltaTable.forPath(spark, "/data/lake/silver/sensors/")
# Z-ordering : clustering intelligent sur les colonnes de filtre
delta_table.optimize() \
.executeZOrderBy("machine_id", "date")
# Compaction : fusionner les petits fichiers
delta_table.optimize() \
.executeCompaction()
# Vacuum : supprimer les fichiers non référencés (après la période de rétention)
delta_table.vacuum(retentionHours=168) # 7 jours
2.6 Delta Change Data Feed
# Activer le CDF
spark.sql("""
ALTER TABLE delta.`/data/lake/silver/sensors/`
SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")
# Lire les changements depuis la version 5
df_changes = spark.read.format("delta") \
.option("readChangeDataFeed", "true") \
.option("startingVersion", "5") \
.load("/data/lake/silver/sensors/")
-- Delta Lake via TrinoCREATE TABLE delta_sensors (
machine_id VARCHAR,
temperature DOUBLE,
pression DOUBLE,
timestampTIMESTAMP
) WITH (
format ='PARQUET',
external_location ='s3://data-lake/delta/sensors/'
);
-- Iceberg via TrinoSELECT*FROM iceberg.sensors.sensor_readings
WHERE machine_id ='M-001'ANDtimestamp>=TIMESTAMP'2026-07-20';
6.2 DuckDB (Analyse Locale Rapide)
-- Lire directement depuis Delta/Iceberg
INSTALL delta;
LOAD delta;
SELECT machine_id, AVG(temperature) AS avg_temp
FROM delta_scan('s3://data-lake/delta/sensors/')
WHERE machine_id IN ('M-001', 'M-002')
GROUPBY machine_id;
-- Iceberg via DuckDB
INSTALL iceberg;
LOAD iceberg;
SELECT*FROM iceberg_scan('s3://data-lake/iceberg/sensors/');
Pièges Courants (Pitfalls)
Petits fichiers (small file problem).
Erreur : Des milliers de fichiers de 1 Mo → métadonnées énormes, lectures lentes.
