Charger des données avec fusion, insertion et ajout

Après avoir transformé et nettoyé les données, la dernière étape de votre pipeline consiste à la charger dans des tables cibles. Azure Databricks fournit trois stratégies de chargement principales : ajout, remplacement et fusion. Chaque stratégie sert différents cas d’usage et le choix de celui qui convient dépend de la façon dont vos données sources sont liées aux enregistrements existants.

Dans cette unité, vous allez apprendre à ajouter de nouvelles lignes avec INSERT, remplacer des données par des opérations de remplacement et synchroniser les modifications à l’aide de MERGE pour les scénarios upsert.

Ajouter des données avec INSERT INTO

L’opération de chargement la plus simple ajoute de nouvelles lignes à une table existante sans modifier les données actuelles. Utilisez INSERT INTO lorsque vous chargez de nouveaux lots de données qui ne chevauchent pas les enregistrements existants, tels que les journaux des transactions quotidiennes, les nouveaux flux d’événements ou les extractions de données incrémentielles.

INSERT INTO fonctionne avec SQL et PySpark. Les exemples suivants montrent comment ajouter des lignes à une table des ventes.

-- Append new records using VALUES
INSERT INTO sales.transactions (transaction_id, amount, transaction_date)
VALUES 
    ('TXN001', 150.00, '2026-01-15'),
    ('TXN002', 275.50, '2026-01-15');

-- Append from a staging table or query
INSERT INTO sales.transactions
SELECT * FROM staging.daily_transactions
WHERE transaction_date = current_date();

Dans PySpark, vous ajoutez des données en écrivant un DataFrame en append mode :

# Append DataFrame to existing table
df_new_transactions.write.mode("append").saveAsTable("sales.transactions")

# Or use insertInto for existing tables
df_new_transactions.write.insertInto("sales.transactions")

Remplacer des données par INSERT OVERWRITE

Parfois, vous devez remplacer les données existantes plutôt que de les ajouter. INSERT OVERWRITE tronque la table ou la partition cible avant d’insérer de nouvelles lignes. Cette approche fonctionne bien pour le retraitement des lots ayant échoué, l’actualisation des tables de recherche ou la reconstruction des agrégations.

-- Replace entire table contents
INSERT OVERWRITE sales.daily_summary
SELECT 
    region,
    SUM(amount) as total_sales,
    COUNT(*) as transaction_count
FROM sales.transactions
WHERE transaction_date = current_date()
GROUP BY region;

Pour les tables partitionnée, vous pouvez remplacer des partitions spécifiques tout en laissant les autres intactes :

-- Replace only January 2026 partition
INSERT OVERWRITE sales.monthly_report
PARTITION (report_month = '2026-01')
SELECT region, product_category, SUM(amount) as revenue
FROM sales.transactions
WHERE transaction_date BETWEEN '2026-01-01' AND '2026-01-31'
GROUP BY region, product_category;

Delta Lake prend également en charge les remplacements sélectifs avec la clause REPLACE WHERE, qui supprime les lignes correspondant à une condition avant d'insérer :

-- Replace transactions for a specific date range
INSERT INTO sales.transactions
REPLACE WHERE transaction_date >= '2026-01-01' AND transaction_date <= '2026-01-07'
SELECT * FROM staging.corrected_transactions;

Par défaut, chaque ligne retournée par la requête doit correspondre au prédicat REPLACE WHERE. Si une seule ligne se trouve en dehors du prédicat, l’instruction échoue et n’écrit rien.

Dans PySpark, utilisez le mode overwrite pour un remplacement complet de la table, et ajoutez l’option replaceWhere pour remplacer uniquement les lignes qui correspondent à une condition :

# Overwrite entire table
df_refreshed.write.mode("overwrite").saveAsTable("sales.daily_summary")

# Replace only the January 2026 rows
(df_january.write
    .mode("overwrite")
    .option("replaceWhere", "report_month = '2026-01'")
    .saveAsTable("sales.monthly_report"))

Remplacer dynamiquement les données par REPLACE USING

REPLACE USING est l’approche recommandée pour les remplacements de données dynamiques. Il fonctionne sur tous les types de calcul, y compris les entrepôts Databricks SQL et le calcul serverless, et ne nécessite pas de configuration de session Spark. Contrairement à partitionOverwriteMode, REPLACE USING prend en charge les tables partitionnées, les tables non partitionnées et les tables avec clustering liquide.

-- Replace only the partitions touched by the incoming data
INSERT INTO sales.monthly_report
  REPLACE USING (report_month)
  SELECT region, product_category, SUM(amount) AS revenue, report_month
  FROM staging.corrected_transactions
  GROUP BY region, product_category, report_month;

La clause REPLACE USING supprime atomiquement les lignes dans sales.monthly_report qui correspondent aux lignes entrantes sur report_month, puis insère les nouvelles lignes. Les lignes qui ne correspondent pas aux colonnes spécifiées sont laissées intactes. Dans Databricks Runtime 16.3 à 17.1, les USING colonnes doivent être l’ensemble complet de colonnes de partition de la table (comportement hérité). Dans Databricks Runtime 17.2 et versions ultérieures, toutes les colonnes sur n’importe quel type de table, y compris les tables non partitionnées et les tables avec Clustering Liquide, sont prises en charge. La clause SQL REPLACE USING nécessite Databricks Runtime 16.3 ou version ultérieure, et l’option Python et Scala replaceUsing nécessite Databricks Runtime 18.2 ou version ultérieure. Étant donné que REPLACE USING utilise l’égalité régulière, les lignes avec des valeurs NULL dans les colonnes USING ne correspondent pas et ne sont pas supprimées de la table cible.

Lorsque vous avez besoin d’une correspondance plus flexible, comme remplacer des lignes en fonction d’une condition personnalisée ou traiter des valeurs NULL comme égales, utilisez REPLACE ON à la place :

-- Replace rows matching a user-defined condition across source and target
INSERT INTO sales.transactions AS target
  REPLACE ON (target.transaction_id = source.transaction_id)
  (SELECT * FROM staging.corrected_transactions) AS source;

REPLACE ON nécessite Databricks Runtime 17.1 et versions ultérieures pour SQL, et Databricks Runtime 18.2 et versions ultérieures pour l’option Python et Scala replaceOn. Utilisez REPLACE WHERE (couvert précédemment) lorsque vous avez besoin d’un remplacement basé sur des conditions sans référence de table source.

Fusionner des données pour les opérations de mise à jour ou d'insertion

Les pipelines de données réels doivent souvent mettre à jour les enregistrements existants lors de l’insertion de nouveaux enregistrements. L’instruction MERGE gère ce modèle, communément appelé « upsert », en faisant correspondre les lignes sources aux lignes cibles et en appliquant différentes actions en fonction de l’existence d’une correspondance.

MERGE est essentiel pour les dimensions à variation lente, la synchronisation des données à partir de systèmes sources et le traitement des flux de capture de données modifiées (CDC).

MERGE INTO customers AS target
USING customer_updates AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
    UPDATE SET 
        target.email = source.email,
        target.phone = source.phone,
        target.last_updated = current_timestamp()
WHEN NOT MATCHED THEN
    INSERT (customer_id, name, email, phone, created_date, last_updated)
    VALUES (source.customer_id, source.name, source.email, source.phone, 
            current_timestamp(), current_timestamp());

Cette requête met à jour les e-mails et les téléphones pour les clients existants tout en insérant des enregistrements clients entièrement nouveaux. La clause ON définit la clé correspondante, généralement une clé primaire ou un identificateur d’entreprise.

Fusionner avec une logique conditionnelle

Ajoutez des conditions aux clauses WHEN pour gérer des scénarios plus complexes. Par exemple, vous pouvez uniquement mettre à jour les enregistrements qui ont réellement changé :

MERGE INTO products AS target
USING product_feed AS source
ON target.sku = source.sku
WHEN MATCHED AND source.price <> target.price THEN
    UPDATE SET target.price = source.price, target.updated_at = current_timestamp()
WHEN MATCHED AND source.discontinued = true THEN
    DELETE
WHEN NOT MATCHED THEN
    INSERT *;

La INSERT * syntaxe insère toutes les colonnes de la source lorsqu’elles correspondent au schéma cible.

Fusionner avec PySpark

Pour le contrôle programmatique, utilisez l’API Delta Lake Python :

from delta.tables import DeltaTable

target_table = DeltaTable.forName(spark, "customers")

target_table.alias("target").merge(
    source_df.alias("source"),
    "target.customer_id = source.customer_id"
).whenMatchedUpdate(set={
    "email": "source.email",
    "phone": "source.phone",
    "last_updated": "current_timestamp()"
}).whenNotMatchedInsert(values={
    "customer_id": "source.customer_id",
    "name": "source.name",
    "email": "source.email",
    "phone": "source.phone",
    "created_date": "current_timestamp()",
    "last_updated": "current_timestamp()"
}).execute()

Pour les cas plus simples où les schémas source et cible correspondent, utilisez les méthodes pratiques :

target_table.alias("target").merge(
    source_df.alias("source"),
    "target.customer_id = source.customer_id"
).whenMatchedUpdateAll(
).whenNotMatchedInsertAll(
).execute()

Choisir la stratégie de chargement appropriée

Votre choix d’opération de chargement dépend de la relation entre les données sources et cibles :

La compréhension de ces modèles de chargement vous prépare à créer des pipelines de données fiables qui conservent l’intégrité des données tout en traitant efficacement les enregistrements nouveaux et modifiés.