Microsoft Fabric Lakehouse :
10 Best Practices pour le Data Engineering

# Verifier si V-Order est actif dans la session
vorder_status = spark.conf.get('spark.sql.parquet.vorder.default')
print(f"V-Order actif : {vorder_status}")

# Activer V-Order au niveau session (tous les writes seront V-Ordered)
spark.conf.set('spark.sql.parquet.vorder.default', 'true')

# Ou activer au niveau table (plus granulaire)
spark.sql("""
    ALTER TABLE gold.dim_customers
    SET TBLPROPERTIES("delta.parquet.vorder.enabled" = "true")
""")

# Ecriture ponctuelle avec V-Order, sans changer la config session
df_report.write \
    .format("delta") \
    .mode("overwrite") \
    .option("parquet.vorder.enabled", "true") \
    .saveAsTable("gold.report_monthly_sales")

# Verifier le statut (devrait etre True dans Fabric)
spark.conf.get('spark.databricks.delta.optimizeWrite.enabled')

# Desactiver temporairement si necessaire (rare)
spark.conf.set('spark.databricks.delta.optimizeWrite.enabled', 'false')

# Ou controller par ecriture individuelle
df.write \
    .format("delta") \
    .mode("append") \
    .option("optimizeWrite", "false") \
    .saveAsTable("bronze.raw_events")

-- Optimiser une table complete
OPTIMIZE silver.fact_transactions;

-- Optimiser avec un filtre (plus rapide, cible les partitions recentes)
OPTIMIZE silver.fact_transactions
WHERE ingestion_date >= '2026-02-01';

-- Optimiser + appliquer V-Order en meme temps
OPTIMIZE gold.dim_products VORDER;

-- Supprimer les fichiers de plus de 7 jours (defaut)
VACUUM silver.fact_transactions;

-- Specifier une retention personnalisee (en heures)
VACUUM silver.fact_transactions RETAIN 168 HOURS;

-- Z-ORDER sur une colonne de filtre frequente
OPTIMIZE silver.fact_transactions
ZORDER BY (customer_id);

-- Z-ORDER multi-colonnes + V-Order combines
OPTIMIZE silver.fact_transactions
WHERE ingestion_date >= '2026-02-01'
ZORDER BY (customer_id, product_id) VORDER;

# ─── BRONZE : Ingestion brute depuis un CSV ───
df_raw = spark.read \
    .format("csv") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .load("Files/raw/sales_2026.csv")

df_raw.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("bronze.raw_sales")

# ─── SILVER : Nettoyage et typage ───
df_silver = spark.sql("""
    SELECT
        CAST(sale_id AS INT) AS sale_id,
        CAST(customer_id AS INT) AS customer_id,
        CAST(amount AS DECIMAL(10,2)) AS amount,
        TO_DATE(sale_date, 'yyyy-MM-dd') AS sale_date,
        current_timestamp() AS processed_at
    FROM bronze.raw_sales
    WHERE amount IS NOT NULL AND amount > 0
""")

df_silver.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("silver.clean_sales")

# ─── GOLD : Agregation metier avec V-Order ───
df_gold = spark.sql("""
    SELECT
        customer_id,
        COUNT(*) AS total_orders,
        SUM(amount) AS total_revenue,
        AVG(amount) AS avg_order_value,
        MAX(sale_date) AS last_purchase_date
    FROM silver.clean_sales
    GROUP BY customer_id
""")

df_gold.write \
    .format("delta") \
    .mode("overwrite") \
    .option("parquet.vorder.enabled", "true") \
    .saveAsTable("gold.customer_metrics")

# Mode APPEND : ajout incremental (Bronze)
df_new_data.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("bronze.events")

# Mode OVERWRITE : remplacement complet (Gold / petites tables)
df_aggregated.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("gold.daily_kpis")

# Mode OVERWRITE partiel avec replaceWhere (Silver / partitions)
df_today.write \
    .format("delta") \
    .mode("overwrite") \
    .option("replaceWhere", "sale_date = '2026-02-25'") \
    .saveAsTable("silver.daily_sales")

from delta.tables import DeltaTable

# Table cible existante
target = DeltaTable.forName(spark, "silver.dim_customers")

# Source : nouvelles donnees / mises a jour
source = df_new_customers

# MERGE : upsert avec Low Shuffle Merge (actif par defaut)
target.alias("t").merge(
    source.alias("s"),
    "t.customer_id = s.customer_id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

# ═══ Notebook de maintenance automatique ═══
# A orchestrer quotidiennement via Data Pipeline

tables_to_maintain = [
    {"name": "silver.fact_transactions", "zorder": ["customer_id"]},
    {"name": "silver.fact_events",       "zorder": ["event_type"]},
    {"name": "gold.customer_metrics",   "zorder": []},
    {"name": "gold.daily_kpis",         "zorder": []},
]

for table in tables_to_maintain:
    name = table["name"]
    zorder_cols = table["zorder"]

    print(f"Maintenance : {name}")

    # Etape 1 : OPTIMIZE (+ Z-ORDER si applicable)
    if zorder_cols:
        cols = ", ".join(zorder_cols)
        spark.sql(f"OPTIMIZE {name} ZORDER BY ({cols}) VORDER")
    else:
        spark.sql(f"OPTIMIZE {name} VORDER")

    # Etape 2 : VACUUM (retention 7 jours par defaut)
    spark.sql(f"VACUUM {name}")

    print(f"  Done !")

print("Maintenance terminee.")