Partager via

Upsert dans une table Delta Lake à l’aide de la fusion

  • Article

Vous pouvez effectuer un upsert de données à partir d’une source table, d’un affichage ou d’une DataFrame dans une table Delta cible à l’aide de l’opération SQL MERGE. Delta Lake prend en charge les insertions, les mises à jour et les suppressions dans MERGE, et prend en charge une syntaxe étendue par rapport aux standards SQL pour faciliter les cas d’usage avancés.

Supposons que vous disposez d'un source table nommée people10mupdates ou d'un chemin source au /tmp/delta/people-10m-updates qui contient de nouvelles données pour un cible table nommé people10m ou un chemin cible au /tmp/delta/people-10m. Certains de ces nouveaux enregistrements sont peut-être déjà présents dans les données cibles. Pour fusionner les nouvelles données, vous devez update les lignes where l’id de la personne est déjà présent, et insert les nouvelles lignes là where aucun id correspondant n’est présent. Vous pouvez exécuter la requête suivante :

SQL

MERGE INTO people10m
USING people10mupdates
ON people10m.id = people10mupdates.id
WHEN MATCHED THEN
  UPDATE SET
    id = people10mupdates.id,
    firstName = people10mupdates.firstName,
    middleName = people10mupdates.middleName,
    lastName = people10mupdates.lastName,
    gender = people10mupdates.gender,
    birthDate = people10mupdates.birthDate,
    ssn = people10mupdates.ssn,
    salary = people10mupdates.salary
WHEN NOT MATCHED
  THEN INSERT (
    id,
    firstName,
    middleName,
    lastName,
    gender,
    birthDate,
    ssn,
    salary
  )
  VALUES (
    people10mupdates.id,
    people10mupdates.firstName,
    people10mupdates.middleName,
    people10mupdates.lastName,
    people10mupdates.gender,
    people10mupdates.birthDate,
    people10mupdates.ssn,
    people10mupdates.salary
  )

Python

from delta.tables import *

deltaTablePeople = DeltaTable.forName(spark, "people10m")
deltaTablePeopleUpdates = DeltaTable.forName(spark, "people10mupdates")

dfUpdates = deltaTablePeopleUpdates.toDF()

deltaTablePeople.alias('people') \
  .merge(
    dfUpdates.alias('updates'),
    'people.id = updates.id'
  ) \
  .whenMatchedUpdate(set =
    {
      "id": "updates.id",
      "firstName": "updates.firstName",
      "middleName": "updates.middleName",
      "lastName": "updates.lastName",
      "gender": "updates.gender",
      "birthDate": "updates.birthDate",
      "ssn": "updates.ssn",
      "salary": "updates.salary"
    }
  ) \
  .whenNotMatchedInsert(values =
    {
      "id": "updates.id",
      "firstName": "updates.firstName",
      "middleName": "updates.middleName",
      "lastName": "updates.lastName",
      "gender": "updates.gender",
      "birthDate": "updates.birthDate",
      "ssn": "updates.ssn",
      "salary": "updates.salary"
    }
  ) \
  .execute()

Scala

import io.delta.tables._
import org.apache.spark.sql.functions._

val deltaTablePeople = DeltaTable.forName(spark, "people10m")
val deltaTablePeopleUpdates = DeltaTable.forName(spark, "people10mupdates")
val dfUpdates = deltaTablePeopleUpdates.toDF()

deltaTablePeople
  .as("people")
  .merge(
    dfUpdates.as("updates"),
    "people.id = updates.id")
  .whenMatched
  .updateExpr(
    Map(
      "id" -> "updates.id",
      "firstName" -> "updates.firstName",
      "middleName" -> "updates.middleName",
      "lastName" -> "updates.lastName",
      "gender" -> "updates.gender",
      "birthDate" -> "updates.birthDate",
      "ssn" -> "updates.ssn",
      "salary" -> "updates.salary"
    ))
  .whenNotMatched
  .insertExpr(
    Map(
      "id" -> "updates.id",
      "firstName" -> "updates.firstName",
      "middleName" -> "updates.middleName",
      "lastName" -> "updates.lastName",
      "gender" -> "updates.gender",
      "birthDate" -> "updates.birthDate",
      "ssn" -> "updates.ssn",
      "salary" -> "updates.salary"
    ))
  .execute()

Important

Une seule ligne de la source table peut correspondre à une ligne donnée dans le tablecible. Dans Databricks Runtime 16.0 et versions ultérieures, MERGE évalue les conditions spécifiées dans les clauses et WHEN MATCHED les ON conditions pour déterminer les correspondances dupliquées. Dans Databricks Runtime 15.4 LTS et ci-dessous, MERGE les opérations considèrent uniquement les conditions spécifiées dans la ON clause.

Pour plus d’informations sur les syntaxes Scala et Python, consultez Documentation sur l’API Delta Lake. Pour plus de détails sur la syntaxe SQL, consultez MERGE INTO

Modifier toutes les lignes sans correspondance à l’aide de la fusion

Dans Databricks SQL et Databricks Runtime 12.2 LTS et versions ultérieures, vous pouvez utiliser la clause WHEN NOT MATCHED BY SOURCE pour les enregistrements UPDATE ou DELETE dans la table cible qui n’a aucun enregistrement correspondant dans la table source. Databricks recommande d’ajouter une clause conditionnelle facultative pour éviter la réécriture totale de l'élément cible table.

L’exemple de code suivant montre la syntaxe de base de l’utilisation de ceci pour les suppressions, en remplaçant le table cible par le contenu du table source et en supprimant les enregistrements non conformes dans le tablecible. Pour utiliser un modèle davantage scalable pour les tables where les mises à jour et les suppressions de la source sont limitées dans le temps, consultez sync de manière incrémentielle la table Delta avec la source.

Python

(targetDF
  .merge(sourceDF, "source.key = target.key")
  .whenMatchedUpdateAll()
  .whenNotMatchedInsertAll()
  .whenNotMatchedBySourceDelete()
  .execute()
)

Scala

targetDF
  .merge(sourceDF, "source.key = target.key")
  .whenMatched()
  .updateAll()
  .whenNotMatched()
  .insertAll()
  .whenNotMatchedBySource()
  .delete()
  .execute()

SQL

MERGE INTO target
USING source
ON source.key = target.key
WHEN MATCHED THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *
WHEN NOT MATCHED BY SOURCE THEN
  DELETE

L’exemple suivant ajoute des conditions à la clause WHEN NOT MATCHED BY SOURCE et spécifie values à update dans des lignes cibles sans correspondance.

Python

(targetDF
  .merge(sourceDF, "source.key = target.key")
  .whenMatchedUpdate(
    set = {"target.lastSeen": "source.timestamp"}
  )
  .whenNotMatchedInsert(
    values = {
      "target.key": "source.key",
      "target.lastSeen": "source.timestamp",
      "target.status": "'active'"
    }
  )
  .whenNotMatchedBySourceUpdate(
    condition="target.lastSeen >= (current_date() - INTERVAL '5' DAY)",
    set = {"target.status": "'inactive'"}
  )
  .execute()
)

Scala

targetDF
  .merge(sourceDF, "source.key = target.key")
  .whenMatched()
  .updateExpr(Map("target.lastSeen" -> "source.timestamp"))
  .whenNotMatched()
  .insertExpr(Map(
    "target.key" -> "source.key",
    "target.lastSeen" -> "source.timestamp",
    "target.status" -> "'active'",
    )
  )
  .whenNotMatchedBySource("target.lastSeen >= (current_date() - INTERVAL '5' DAY)")
  .updateExpr(Map("target.status" -> "'inactive'"))
  .execute()

SQL

MERGE INTO target
USING source
ON source.key = target.key
WHEN MATCHED THEN
  UPDATE SET target.lastSeen = source.timestamp
WHEN NOT MATCHED THEN
  INSERT (key, lastSeen, status) VALUES (source.key,  source.timestamp, 'active')
WHEN NOT MATCHED BY SOURCE AND target.lastSeen >= (current_date() - INTERVAL '5' DAY) THEN
  UPDATE SET target.status = 'inactive'

Sémantique de l’opération de fusion

Vous trouverez ci-dessous une description détaillée de la sémantique de l’opération merge programmatique.

  • Il peut y avoir un nombre quelconque de clauses whenMatched et whenNotMatched.

  • Les clauses whenMatched sont exécutées lorsqu’une ligne source correspond à une ligne de table cible basée sur la condition de correspondance. Ces clauses ont la sémantique suivante.

    • Les clauses whenMatched ne peuvent avoir qu’une action update et une delete. L’action update dans merge met uniquement à jour le columns spécifié (similaire à l’opération de update) de la ligne cible correspondante. L’action delete supprime la ligne correspondante.

    • Chaque clause whenMatched peut avoir une condition facultative. Si cette condition de clause existe, l’action update ou delete est exécutée pour toute paire de lignes source-cible correspondante uniquement lorsque la condition de clause a la valeur true.

    • S’il y a plusieurs clauses whenMatched, elles sont évaluées dans l’ordre dans lequel elles sont spécifiées. Toutes les clauses whenMatched, à l’exception de la dernière, doivent avoir des conditions.

    • Si aucune des conditions whenMatched ne prend la valeur true pour une paire de lignes source et cible qui correspond à la condition de fusion, la ligne cible reste inchangée.

    • Pour update tous les columns de la table Delta cible avec le columns correspondant du jeu de données source, utilisez whenMatched(...).updateAll(). Ceci équivaut à :

      whenMatched(...).updateExpr(Map("col1" -> "source.col1", "col2" -> "source.col2", ...))

      pour toutes les columns de la table Delta cible. Par conséquent, cette action suppose que la table source a les mêmes columns que celles de la tablecible ; sinon, la requête génère une erreur d'analyse.

      Remarque

      Ce comportement change lorsque l’évolution automatique schema est activée. Consultez évolution automatique schema pour plus d’informations.

  • Les clauses whenNotMatched sont exécutées lorsqu’une ligne source ne correspond à aucune ligne cible basée sur la condition de correspondance. Ces clauses ont la sémantique suivante.

    • Les clauses whenNotMatched ne peuvent avoir que l’action insert. La nouvelle ligne est générée en fonction des expressions column et correspondantes spécifiées. Il n’est pas nécessaire de spécifier toutes les columns de la table cible. Pour les cibles columnsnon spécifiées, NULL est inséré.

    • Chaque clause whenNotMatched peut avoir une condition facultative. Si la condition de clause est présente, une ligne source est insérée uniquement si cette condition est vraie pour cette ligne. Sinon, la source column est ignorée.

    • S’il y a plusieurs clauses whenNotMatched, elles sont évaluées dans l’ordre dans lequel elles sont spécifiées. Toutes les clauses whenNotMatched, à l’exception de la dernière, doivent avoir des conditions.

    • Pour insert tous les columns de la table Delta cible avec le columns correspondant du jeu de données source, utilisez whenNotMatched(...).insertAll(). Ceci équivaut à :

      whenNotMatched(...).insertExpr(Map("col1" -> "source.col1", "col2" -> "source.col2", ...))

      pour toutes les columns de la table Delta cible. Par conséquent, cette action suppose que la table source possède les mêmes columns que celles de la table cible, sinon la requête génère une erreur d’analyse.

      Remarque

      Ce comportement change lorsque l’évolution automatique schema est activée. Pour plus d’informations, consultez Évolution automatique du schema.

  • Les clauses whenNotMatchedBySource sont exécutées quand une ligne cible ne correspond à aucune ligne source sur la base de la condition de fusion. Ces clauses ont la sémantique suivante.

    • Les clauses whenNotMatchedBySource peuvent spécifier les actions delete et update.
    • Chaque clause whenNotMatchedBySource peut avoir une condition facultative. Si la condition de clause est présente, une ligne cible est modifiée uniquement si cette condition est vraie pour cette ligne. Sinon, la ligne cible reste inchangée.
    • S’il y a plusieurs clauses whenNotMatchedBySource, elles sont évaluées dans l’ordre dans lequel elles sont spécifiées. Toutes les clauses whenNotMatchedBySource, à l’exception de la dernière, doivent avoir des conditions.
    • Par définition, les clauses whenNotMatchedBySource n’ont pas de ligne source à partir de laquelle extraire columnvalues, et les columns sources ne peuvent donc pas être référencées. Pour chaque column à modifier, vous pouvez spécifier une valeur littérale ou effectuer une action sur la cible column, telle que SET target.deleted_count = target.deleted_count + 1.

Important

  • Une opération merge peut échouer si plusieurs lignes du jeu de données source correspondent et si l’opération tente de update les mêmes lignes de la table Delta cible. Selon la sémantique SQL de fusion, une telle opération de update est ambiguë, car il n’est pas clair quelle ligne source doit être utilisée pour update la ligne cible correspondante. Vous pouvez prétraiter la source table pour éliminer la possibilité de plusieurs correspondances.
  • Vous pouvez appliquer une opération SQL MERGE sur un affichage SQL VIEW uniquement si l’affichage a été défini en tant que CREATE VIEW viewName AS SELECT * FROM deltaTable.

La déduplication des données lors de l’écriture dans Delta tables

Un cas d’usage ETL courant consiste à collecter des journaux dans une table Delta en les ajoutant à une table. Toutefois, les sources peuvent souvent generate des enregistrements de journal en double et des étapes de déduplication en aval sont nécessaires pour s’en occuper. La commande merge vous permet d’éviter d’insérer les enregistrements en double.

SQL

MERGE INTO logs
USING newDedupedLogs
ON logs.uniqueId = newDedupedLogs.uniqueId
WHEN NOT MATCHED
  THEN INSERT *

Python

deltaTable.alias("logs").merge(
    newDedupedLogs.alias("newDedupedLogs"),
    "logs.uniqueId = newDedupedLogs.uniqueId") \
  .whenNotMatchedInsertAll() \
  .execute()

Scala

deltaTable
  .as("logs")
  .merge(
    newDedupedLogs.as("newDedupedLogs"),
    "logs.uniqueId = newDedupedLogs.uniqueId")
  .whenNotMatched()
  .insertAll()
  .execute()

Java

deltaTable
  .as("logs")
  .merge(
    newDedupedLogs.as("newDedupedLogs"),
    "logs.uniqueId = newDedupedLogs.uniqueId")
  .whenNotMatched()
  .insertAll()
  .execute();

Remarque

Le jeu de données contenant les nouveaux journaux doit être dédupliqué dans son propre sein. Par la sémantique SQL de fusion, elle correspond et déduplique les nouvelles données avec les données existantes dans l'table, mais s’il existe des données en double dans le nouveau jeu de données, elle est insérée. Par conséquent, dédupliquez les nouvelles données avant de les fusionner dans la table.

Si vous savez que vous risquez d’get des enregistrements en double pendant quelques jours seulement, vous pouvez optimize votre requête en partitionnant la table par date, puis en spécifiant la plage de dates de la table cible avec laquelle établir la correspondance.

SQL

MERGE INTO logs
USING newDedupedLogs
ON logs.uniqueId = newDedupedLogs.uniqueId AND logs.date > current_date() - INTERVAL 7 DAYS
WHEN NOT MATCHED AND newDedupedLogs.date > current_date() - INTERVAL 7 DAYS
  THEN INSERT *

Python

deltaTable.alias("logs").merge(
    newDedupedLogs.alias("newDedupedLogs"),
    "logs.uniqueId = newDedupedLogs.uniqueId AND logs.date > current_date() - INTERVAL 7 DAYS") \
  .whenNotMatchedInsertAll("newDedupedLogs.date > current_date() - INTERVAL 7 DAYS") \
  .execute()

Scala

deltaTable.as("logs").merge(
    newDedupedLogs.as("newDedupedLogs"),
    "logs.uniqueId = newDedupedLogs.uniqueId AND logs.date > current_date() - INTERVAL 7 DAYS")
  .whenNotMatched("newDedupedLogs.date > current_date() - INTERVAL 7 DAYS")
  .insertAll()
  .execute()

Java

deltaTable.as("logs").merge(
    newDedupedLogs.as("newDedupedLogs"),
    "logs.uniqueId = newDedupedLogs.uniqueId AND logs.date > current_date() - INTERVAL 7 DAYS")
  .whenNotMatched("newDedupedLogs.date > current_date() - INTERVAL 7 DAYS")
  .insertAll()
  .execute();

Cela est plus efficace que la commande précédente, car elle recherche des doublons uniquement dans les 7 derniers jours de logs, et non dans l'ensemble de la table. En outre, vous pouvez utiliser cette fusion par insert uniquement avec une diffusion en continu structurée afin d’effectuer une déduplication continue des journaux.

  • Dans une requête de diffusion en continu, vous pouvez utiliser une opération de fusion dans foreachBatch afin d’écrire en permanence toutes les données de diffusion en continu dans une table Delta avec déduplication. Pour plus d’informations sur , consultez l'foreachBatch suivant.
  • Dans une autre requête de diffusion en continu, vous pouvez lire en permanence des données dédupliquées à partir de cette table Delta. Cela est possible parce qu’une fusion par insert uniquement ajoute uniquement de nouvelles données à la table Delta.

Données à variation lente (SCD) et capture de données modifiées (CDC) avec Delta Lake

Delta Live Tables prend en charge nativement le suivi et l’application de SCD Type 1 et Type 2. Utilisez APPLY CHANGES INTO avec Delta Live Tables pour vous assurer que les enregistrements en désordre sont gérés correctement lors du traitement des flux CDC. Consultez les API APPLY CHANGES : Simplifier la capture des changements de données avec Delta Live Tables.

sync de manière incrémentielle la table Delta avec la source

Dans Databricks SQL et Databricks Runtime 12.2 LTS et versions ultérieures, vous pouvez utiliser WHEN NOT MATCHED BY SOURCE pour créer des conditions arbitraires pour supprimer et remplacer atomiquement une partie d’une table. Cela peut être particulièrement utile lorsque vous disposez d'une source où les enregistrements tablewhere peuvent changer ou être supprimés pendant plusieurs jours après l'entrée de données initiale, mais finissent par se stabiliser dans un état final.

La requête suivante montre l’utilisation de ce modèle pour select 5 jours d’enregistrements de la source, update les enregistrements correspondants dans la cible, insert nouveaux enregistrements de la source vers la cible et supprimer tous les enregistrements non compatibles des 5 derniers jours dans la cible.

MERGE INTO target AS t
USING (SELECT * FROM source WHERE created_at >= (current_date() - INTERVAL '5' DAY)) AS s
ON t.key = s.key
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
WHEN NOT MATCHED BY SOURCE AND created_at >= (current_date() - INTERVAL '5' DAY) THEN DELETE

En fournissant le même filtre booléen sur la source tableset la cible tables, vous pouvez propager dynamiquement les modifications de votre source vers votre cible tables, y compris les suppressions.

Remarque

Bien que ce modèle puisse être utilisé sans clauses conditionnelles, cela entraînerait une réécriture complète de la cible table qui peut être coûteuse.