Upsert dans une table Delta Lake à l’aide de la fusion
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
etwhenNotMatched
.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 actionupdate
et unedelete
. L’actionupdate
dansmerge
met uniquement à jour le columns spécifié (similaire à l’opération deupdate
) de la ligne cible correspondante. L’actiondelete
supprime la ligne correspondante.Chaque clause
whenMatched
peut avoir une condition facultative. Si cette condition de clause existe, l’actionupdate
oudelete
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 clauseswhenMatched
, à 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’actioninsert
. 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 clauseswhenNotMatched
, à 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 actionsdelete
etupdate
. - 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 clauseswhenNotMatchedBySource
, à 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 queSET target.deleted_count = target.deleted_count + 1
.
- Les clauses
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 queCREATE 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.