# =============================================================================
# CONSOLIDATION MULTI-CLIENTS : Bronze → Silver
# =============================================================================
# Ce notebook fusionne les tables de plusieurs lakehouses Bronze vers un
# lakehouse Silver centralisé. Il gère :
# - la lecture dynamique de la liste des tables depuis un CSV
# - l'ajout d'une colonne client_name sur chaque ligne
# - les différences de schéma entre clients (colonnes absentes → null)
# - trois stratégies d'écriture selon le type de clé :
# * ChampCléDate → overwrite de partition (idempotent, performant)
# * ChampCléID → MERGE Delta (upsert, sans doublon)
# * Aucune clé → overwrite complet
# =============================================================================
from pyspark.sql.functions import lit, col
from pyspark.sql import functions as F
from delta.tables import DeltaTable
import time
# =============================================================================
# 1. CONFIGURATION
# =============================================================================
# Déclarez ici chaque client avec le nom de son lakehouse Bronze.
# Pour ajouter un 4e client, il suffit d'ajouter une entrée dans cette liste.
clients = [
{"client_name": "Client1", "dbo": "dboClient1"},
{"client_name": "Client2", "dbo": "dboClient2"},
{"client_name": "Client3", "dbo": "dboClient3"},
]
# Chemin du CSV dans la section Files du lakehouse Silver (lakehouse par défaut
# du notebook). Le CSV doit contenir trois colonnes :
# NomTable ; ChampCléDate ; ChampCléID
# Une seule des deux colonnes clé est renseignée à la fois (ou aucune).
CHEMIN_CSV = "Files/ListeTable.csv"
print("=== Démarrage de la consolidation multi-clients ===")
print(f" {len(clients)} clients configurés : {[c['client_name'] for c in clients]}")
start_total = time.time()
# =============================================================================
# 2. LECTURE DYNAMIQUE DE LA LISTE DES TABLES
# =============================================================================
print(f"\nLecture de {CHEMIN_CSV}...")
liste_tables_df = (
spark.read
.option("header", "true")
.option("delimiter", ";")
.csv(CHEMIN_CSV)
)
# On construit un dictionnaire de configuration par table.
# Les champs vides du CSV deviennent None pour faciliter les tests booléens.
tables_config = [
{
"NomTable": row["NomTable"],
"ChampCléDate": row["ChampCléDate"] if row["ChampCléDate"] else None,
"ChampCléID": row["ChampCléID"] if row["ChampCléID"] else None,
}
for row in liste_tables_df.collect()
if row["NomTable"] is not None
]
print(f" {len(tables_config)} tables à traiter : {[t['NomTable'] for t in tables_config]}")
# =============================================================================
# 3. FONCTIONS D'ÉCRITURE (une par stratégie)
# =============================================================================
def ecrire_overwrite_complet(df_final, table_silver):
"""
Stratégie pour les tables sans clé : on réécrit toute la table Silver.
Simple et fiable, mais coûteux si la table est volumineuse.
"""
(
df_final.write
.format("delta")
.mode("overwrite")
.option("mergeSchema", "true")
.saveAsTable(table_silver)
)
print(f" Écriture : overwrite complet → {table_silver}")
def ecrire_overwrite_partition(df_final, table_silver, champ_date):
"""
Stratégie recommandée pour ChampCléDate.
On calcule la plage de dates présente dans le nouveau lot Bronze, puis on
réécrit uniquement la ou les partitions concernées dans le Silver.
Idempotent : rejouer le notebook sur la même plage donne le même résultat.
"""
# Calcul de la plage couverte par le nouveau lot
valeur_min = df_final.agg(F.min(col(champ_date))).collect()[0][0]
valeur_max = df_final.agg(F.max(col(champ_date))).collect()[0][0]
if valeur_min is None:
print(f" Colonne {champ_date} entièrement null, basculement en overwrite complet.")
ecrire_overwrite_complet(df_final, table_silver)
return
# Colonne de partition au format YYYY-MM pour regrouper par mois.
# Vous pouvez changer "yyyy-MM" en "yyyy" (année) ou "yyyy-MM-dd" (jour)
# selon le volume quotidien de vos données.
df_final = df_final.withColumn(
"partition_mois",
F.date_format(col(champ_date), "yyyy-MM")
)
mois_min = str(valeur_min)[:7]
mois_max = str(valeur_max)[:7]
print(f" Plage détectée : {valeur_min} → {valeur_max} (partitions {mois_min} à {mois_max})")
(
df_final.write
.format("delta")
.mode("overwrite")
# replaceWhere cible uniquement les partitions de la plage, sans toucher
# aux données historiques en dehors de cette fenêtre.
.option("replaceWhere", f"partition_mois >= '{mois_min}' AND partition_mois <= '{mois_max}'")
.option("mergeSchema", "true")
.partitionBy("partition_mois")
.saveAsTable(table_silver)
)
print(f" Écriture : overwrite partition [{mois_min} → {mois_max}] → {table_silver}")
def ecrire_merge(df_final, table_silver, champ_id):
"""
Stratégie recommandée pour ChampCléID.
Le MERGE Delta compare chaque ligne entrante avec la table Silver sur la
clé ID + client_name. Si la ligne existe déjà, elle est mise à jour ;
sinon elle est insérée. Aucun doublon possible, même en cas de rejeu.
"""
if spark.catalog.tableExists(table_silver):
dt = DeltaTable.forName(spark, table_silver)
(
dt.alias("silver")
.merge(
df_final.alias("bronze"),
# La clé composite (champ_id + client_name) garantit l'unicité
# par client, même si deux clients partagent les mêmes IDs.
f"silver.{champ_id} = bronze.{champ_id} "
f"AND silver.client_name = bronze.client_name"
)
.whenMatchedUpdateAll() # met à jour les lignes modifiées
.whenNotMatchedInsertAll() # insère les nouvelles lignes
.execute()
)
print(f" Écriture : MERGE sur {champ_id} + client_name → {table_silver}")
else:
# Premier chargement : la table Silver n'existe pas encore
print(f" Table Silver absente, premier chargement complet par overwrite.")
ecrire_overwrite_complet(df_final, table_silver)
# =============================================================================
# 4. BOUCLE PRINCIPALE DE CONSOLIDATION
# =============================================================================
tables_ok = 0
tables_en_erreur = []
for i, config in enumerate(tables_config, start=1):
table_name = config["NomTable"]
champ_date = config["ChampCléDate"]
champ_id = config["ChampCléID"]
table_silver = f"{table_name}_unified"
# Détermination du mode pour ce tour de boucle
if champ_date:
mode_label = f"overwrite partition sur {champ_date}"
elif champ_id:
mode_label = f"MERGE sur {champ_id}"
else:
mode_label = "overwrite complet (aucune clé)"
print(f"\n--- [{i}/{len(tables_config)}] {table_name} | {mode_label} ---")
start_table = time.time()
# ------------------------------------------------------------------
# Lecture et filtrage incrémental depuis chaque lakehouse Bronze
# ------------------------------------------------------------------
dfs = []
for client in clients:
try:
df = spark.table(f"{client['dbo']}.{table_name}")
nb_brut = df.count()
# Pour les stratégies incrémentales, on ne lit que les nouvelles
# lignes en calculant d'abord le MAX présent dans le Silver.
champ_cle = champ_date or champ_id
if champ_cle and spark.catalog.tableExists(table_silver):
valeur_max_silver = (
spark.table(table_silver)
.agg(F.max(col(champ_cle)))
.collect()[0][0]
)
if valeur_max_silver is not None:
df = df.filter(col(champ_cle) > valeur_max_silver)
nb_filtre = df.count()
print(f" {client['client_name']} : {nb_brut} lignes totales "
f"→ {nb_filtre} nouvelles ({champ_cle} > {valeur_max_silver})")
else:
print(f" {client['client_name']} : {nb_brut} lignes (Silver vide, chargement complet)")
else:
print(f" {client['client_name']} : {nb_brut} lignes (chargement complet)")
df = df.withColumn("client_name", lit(client["client_name"]))
dfs.append(df)
except Exception as e:
print(f" {client['client_name']} : ERREUR — {e}")
tables_en_erreur.append({
"table": table_name,
"client": client["client_name"],
"erreur": str(e)
})
continue
if not dfs:
print(f" Aucune source disponible, table ignorée.")
continue
# ------------------------------------------------------------------
# Diagnostic des colonnes absentes entre clients
# ------------------------------------------------------------------
tous_colonnes = set.union(*[set(df.columns) for df in dfs])
for df, client in zip(dfs, clients):
manquantes = tous_colonnes - set(df.columns) - {"client_name"}
if manquantes:
print(f" Colonnes absentes chez {client['client_name']} (seront null) : {manquantes}")
# ------------------------------------------------------------------
# Union des dataframes de tous les clients
# ------------------------------------------------------------------
df_final = dfs[0]
for df in dfs[1:]:
df_final = df_final.unionByName(df, allowMissingColumns=True)
nb_final = df_final.count()
print(f" Résultat fusionné : {nb_final} lignes, {len(df_final.columns)} colonnes")
if nb_final == 0:
print(f" Aucune nouvelle ligne à écrire, étape ignorée.")
tables_ok += 1
continue
# ------------------------------------------------------------------
# Écriture selon la stratégie déterminée par le CSV
# ------------------------------------------------------------------
if champ_date:
ecrire_overwrite_partition(df_final, table_silver, champ_date)
elif champ_id:
ecrire_merge(df_final, table_silver, champ_id)
else:
ecrire_overwrite_complet(df_final, table_silver)
duree_table = round(time.time() - start_table, 1)
print(f" Terminé en {duree_table}s")
tables_ok += 1
# =============================================================================
# 5. RÉCAPITULATIF FINAL
# =============================================================================
duree_totale = round(time.time() - start_total, 1)
print(f"\n=== Consolidation terminée en {duree_totale}s ===")
print(f" Tables traitées avec succès : {tables_ok}/{len(tables_config)}")
if tables_en_erreur:
print(f" Erreurs rencontrées ({len(tables_en_erreur)}) :")
for err in tables_en_erreur:
print(f" - {err['table']} / {err['client']} : {err['erreur']}")
else:
print(" Aucune erreur.")
# =============================================================================
# 6. MAINTENANCE (à exécuter dans un notebook séparé, planifié hebdomadairement)
# =============================================================================
# Les opérations OPTIMIZE et VACUUM ne sont pas incluses dans la boucle
# principale pour ne pas ralentir chaque exécution. Les exécuter séparément
# sur un pipeline planifié (ex : tous les dimanches).
#
# for config in tables_config:
# table_silver = f"{config['NomTable']}_unified"
# spark.sql(f"OPTIMIZE {table_silver}")
# spark.sql(f"VACUUM {table_silver} RETAIN 168 HOURS") # 7 jours
# print(f" OPTIMIZE + VACUUM : {table_silver}")
