Écrasement sélectif des données avec Delta Lake

Delta Lake offre les options distinctes suivantes pour les remplacements sélectifs :

Option	Cas d'usage	Types de compute pris en charge	Version minimale
REPLACE WHERE	Écraser atomiquement les lignes qui correspondent à un prédicat. Utilisez pour les remplacements avec une condition de correspondance fixe, tels que colA = 5 ou int_col IN (1, 2, 3).	Tous les types de compute.	SQL dans Databricks Runtime 12,2 LTS et versions ultérieures. Python et Scala dans Databricks Runtime 9.1 LTS et versions ultérieures.
REPLACE USING	Écrasement dynamique des données. Remplace toutes les lignes qui correspondent aux colonnes spécifiées, en fonction de la comparaison d'égalité des valeurs de colonne, dans l'ensemble de données fourni.	Tous les types de compute.	SQL dans Databricks Runtime 16.3 et versions ultérieures. Python et Scala dans Databricks Runtime 18.2 et versions ultérieures.
REPLACE ON	Écrasement dynamique des données par expression booléenne. À utiliser pour des remplacements avec une condition de correspondance complexe ou protégée contre les valeurs nulles, telle que s.colA <=> t.colA AND s.colB <=> t.colB.	Tous les types de compute.	SQL dans Databricks Runtime 17,1 et versions ultérieures. Python et Scala dans Databricks Runtime 18.2 et versions ultérieures.
partitionOverwriteMode	Remplacement dynamique de partition hérité, qui écrase toutes les données existantes dans chaque partition pour laquelle l'écriture va commit de nouvelles données. Non recommandé pour les nouvelles workloads.	SQL prend en charge uniquement le compute classique. Python et Scala prennent en charge tous les types de compute.	SQL, Python et Scala dans Databricks Runtime 11.3 LTS et versions ultérieures.

Option	Cas d'usage	Types de compute pris en charge	Version minimale
REPLACE WHERE	Écraser atomiquement les lignes qui correspondent à un prédicat. Utilisez pour les remplacements avec une condition de correspondance fixe, tels que colA = 5 ou int_col IN (1, 2, 3).	Tous les types de compute.	SQL dans Databricks Runtime 12,2 LTS et versions ultérieures. Python et Scala dans Databricks Runtime 9.1 LTS et versions ultérieures.
REPLACE USING	Écrasement dynamique des données. Remplace toutes les lignes qui correspondent aux colonnes spécifiées, en fonction de la comparaison d'égalité des valeurs de colonne, dans l'ensemble de données fourni.	Tous les types de compute.	SQL dans Databricks Runtime 16.3 et versions ultérieures. Python et Scala dans Databricks Runtime 18.2 et versions ultérieures.
REPLACE ON	Écrasement dynamique des données par expression booléenne. À utiliser pour des remplacements avec une condition de correspondance complexe ou protégée contre les valeurs nulles, telle que s.colA <=> t.colA AND s.colB <=> t.colB.	Tous les types de compute.	SQL dans Databricks Runtime 17,1 et versions ultérieures. Python et Scala dans Databricks Runtime 18.2 et versions ultérieures.
partitionOverwriteMode	Remplacement dynamique de partition hérité, qui écrase toutes les données existantes dans chaque partition pour laquelle l'écriture va commit de nouvelles données. Non recommandé pour les nouvelles workloads.	SQL prend en charge uniquement le compute classique. Python et Scala prennent en charge tous les types de compute.	SQL, Python et Scala dans Databricks Runtime 11.3 LTS et versions ultérieures.

Pour la plupart des cas d'utilisation, Databricks recommande d'utiliser REPLACE USING ou REPLACE WHERE. Utilisez REPLACE ON uniquement si votre cas d'utilisation requiert des conditions de correspondance complexes ou NULL-safe.

Pour plus de détails sur le comportement de remplacement de chaque option, consultez INSERT. Pour une liste complète des options DataFrameWriter Delta Lake, consultez Delta Lake et Apache Iceberg.

En Scala et Python, replaceOn et replaceUsing ne peuvent pas être utilisés en combinaison avec replaceWhere, partitionOverwriteMode ou overwriteSchema.

Pour les queries sources vides, REPLACE USING et REPLACE ON ne suppriment pas de données ; cependant, REPLACE WHERE pourrait en supprimer.

important

Si les données ont été accidentellement écrasées, vous pouvez utiliser restaurer pour annuler la modification.

REPLACE WHERE

Vous pouvez remplacer sélectivement uniquement les données qui correspondent à une expression arbitraire avec REPLACE WHERE.

info

Bêta

Pour bénéficier du refresh incrémentiel lors de l'exécution de REPLACE WHERE, utilisez les flux REPLACE WHERE dans les LakeFlow Pipelines. Voir Traitement par batch avec les flux REPLACE WHERE.

Pour remplacer atomiquement les événements de janvier dans la table cible, qui est partitionnée par start_date, avec les données de replace_data:

Python

Python

(replace_data.write
  .mode("overwrite")
  .option("replaceWhere", "start_date >= '2017-01-01' AND end_date <= '2017-01-31'")
  .saveAsTable("events")
)

Scala

Scala

replace_data.write
  .mode("overwrite")
  .option("replaceWhere", "start_date >= '2017-01-01' AND end_date <= '2017-01-31'")
  .saveAsTable("events")

SQL

SQL

INSERT INTO TABLE events REPLACE WHERE start_date >= '2017-01-01' AND end_date <= '2017-01-31' SELECT * FROM replace_data

Ce code exemple écrit les données dans replace_data, valide que toutes les lignes correspondent au prédicat et effectue un remplacement atomique en utilisant la sémantique overwrite. Si des valeurs dans l'opération se trouvent en dehors du prédicat, cette opération échoue par default avec une erreur.

Sur le compute classique, pour changer ce comportement en overwrite valeurs dans la plage de prédicats et insert enregistrements en dehors de la plage spécifiée, supprimez la vérification des contraintes en définissant spark.databricks.delta.replaceWhere.constraintCheck.enabled sur false:

Python

Python

spark.conf.set("spark.databricks.delta.replaceWhere.constraintCheck.enabled", False)

Scala

Scala

spark.conf.set("spark.databricks.delta.replaceWhere.constraintCheck.enabled", false)

SQL

SQL

SET spark.databricks.delta.replaceWhere.constraintCheck.enabled=false

remarque

REPLACE WHERE accepte un boolean_expression avec quelques restrictions. Voir INSERT dans la référence du langage SQL.

Pour les requêtes sources vides, REPLACE WHERE peut supprimer des lignes de table.

Comportement hérité

Le replaceWhere hérité est uniquement disponible sur le compute classique. Voir Aperçu de Classic compute.

Si vous utilisez le comportement hérité de replaceWhere, les requêtes écrasent les données qui correspondent à un prédicat uniquement sur les colonnes de partition. La commande suivante remplacerait atomiquement le mois de janvier dans la table cible, qui est partitionnée par date, avec les données de df:

Python

Python

(df.write
  .mode("overwrite")
  .option("replaceWhere", "birthDate >= '2017-01-01' AND birthDate <= '2017-01-31'")
  .saveAsTable("people10m")
)

Scala

Scala

df.write
  .mode("overwrite")
  .option("replaceWhere", "birthDate >= '2017-01-01' AND birthDate <= '2017-01-31'")
  .saveAsTable("people10m")

Pour utiliser le comportement hérité, définissez spark.databricks.delta.replaceWhere.dataColumns.enabled sur false:

Python

Python

spark.conf.set("spark.databricks.delta.replaceWhere.dataColumns.enabled", False)

Scala

Scala

spark.conf.set("spark.databricks.delta.replaceWhere.dataColumns.enabled", false)

SQL

SQL

SET spark.databricks.delta.replaceWhere.dataColumns.enabled=false

Écrasements de données dynamiques

Les écrasements de données dynamiques remplacent sélectivement les données qui correspondent aux colonnes clés spécifiées ou à l'expression booléenne, laissant toutes les autres données inchangées. Les tables partitionnées, les tables non partitionnées et les tables avec clustering liquide sont toutes prises en charge.

Les remplacements de partition dynamiques sont un sous-ensemble du comportement de remplacement de données dynamiques. Les écrasements de partitions dynamiques remplacent toutes les données existantes dans chaque partition pour laquelle l'écriture will commit de nouvelles données et laissent toutes les autres partitions inchangées. Seules les tables partitionnées sont prises en charge.

REPLACE USING

SQL pris en charge dans Databricks Runtime 16.3 et versions ultérieures. Python et Scala pris en charge dans Databricks Runtime 18.2 et versions ultérieures. Pour les différences de comportement dans Databricks Runtime de 16.3 à 17.1, consultez Comportement hérité.

REPLACE USING permet un comportement de remplacement atomique et indépendant du compute qui fonctionne sur les Databricks SQL warehouse, le Serverless compute et le compute classique. REPLACE USING ne vous demande pas de définir une configuration de session Spark.

REPLACE USING remplace les lignes lorsque les colonnes spécifiées sont égales. Toutes les autres données restent inchangées.

Utilisez l'écrasement dynamique des données avec REPLACE USING:

Python

Python

(sourceDataDF.write
  .mode("overwrite")
  .option("replaceUsing", "event_id, start_date")
  .saveAsTable("events")
)

Scala

Scala

sourceDataDF.write
  .mode("overwrite")
  .option("replaceUsing", "event_id, start_date")
  .saveAsTable("events")

SQL

SQL

INSERT INTO TABLE events
  REPLACE USING (event_id, start_date)
  SELECT * FROM source_data

Pour les queries source vides, REPLACE USING ne supprime aucune ligne de table.

Pour les conditions de correspondance complexes ou NULL-safe, utilisez REPLACE ON à la place. Voir REPLACE ON.

Voir INSERT dans la référence du langage SQL.

Comportement hérité

Dans Databricks Runtime 16.3 à 17.1, REPLACE USING utilise un comportement hérité et n'autorise que les écrasements de partitions dynamiques, tandis que Databricks Runtime 17.2 et supérieur permet les écrasements de données dynamiques.

Gardez les contraintes et comportements suivants à l'esprit pour le comportement hérité REPLACE USING :

• Vous devez spécifier l'ensemble complet des colonnes de partition de la table dans la clause USING.
• Validez toujours que les données écrites ne touchent que les partitions attendues. Une seule ligne dans la mauvaise partition peut écraser involontairement toute la partition.

REPLACE ON

SQL pris en charge dans Databricks Runtime 17.1 et versions ultérieures. Python et Scala sont pris en charge dans Databricks Runtime 18.2 et versions ultérieures.

REPLACE ON remplace les lignes lorsqu'elles correspondent à une condition définie par l'utilisateur, contrairement à REPLACE USING, qui remplace les lignes lorsque les colonnes spécifiées sont égales. Utilisez REPLACE ON lorsque vous avez besoin d’une logique de correspondance que REPLACE USING ne prend pas en charge, par exemple pour considérer les valeurs NULL comme égales.

Facultativement, utilisez l'option targetAlias pour spécifier un alias pour la table cible et les APIs .as() ou .alias() pour spécifier un alias pour les données sources.

Pour la syntaxe SQL, consultez INSERT.

Python

Python

(sourceDataDF.alias("s")
  .write
  .mode("overwrite")
  .option("targetAlias", "t")
  .option("replaceOn", "s.event_id <=> t.event_id AND s.start_date <=> t.start_date")
  .saveAsTable("events")
)

Scala

Scala

sourceDataDF.as("s")
  .write
  .mode("overwrite")
  .option("targetAlias", "t")
  .option("replaceOn", "s.event_id <=> t.event_id AND s.start_date <=> t.start_date")
  .saveAsTable("events")

SQL

SQL

INSERT INTO TABLE events AS t
  REPLACE ON (s.event_id <=> t.event_id AND s.start_date <=> t.start_date)
  (SELECT * FROM source_data) AS s

Pour les queries source vides, REPLACE ON ne supprime aucune ligne de table.

Écritures écrasées de partitions dynamiques avec partitionOverwriteMode (hérité)

info

Aperçu

Cette fonctionnalité est en aperçu public.

Databricks Runtime 11.3 LTS et versions ultérieures prend en charge les remplacements dynamiques de partitions pour les tables partitionnées en mode de remplacement : soit INSERT OVERWRITE en SQL, soit une écriture de DataFrame avec df.write.mode("overwrite"). Ce type de remplacement n'est disponible que pour le compute classique, pas pour les Databricks SQL warehouse ou le Serverless compute.

attention

Dans la mesure du possible, utilisez INSERT REPLACE USING au lieu de la surcharge de partition INSERT OVERWRITE PARTITION et spark.sql.sources.partitionOverwriteMode=dynamic. L'écrasement de partition peut utiliser des données obsolètes lorsque le partitionnement change.

Pour utiliser le mode de remplacement de partition dynamique, définissez la configuration de session Spark spark.sql.sources.partitionOverwriteMode sur dynamic. Alternativement, vous pouvez définir l’option DataFrameWriter partitionOverwriteMode sur dynamic. Si elle est présente, l'option spécifique à la query remplace le mode défini dans la configuration de la session. La valeur default pour spark.sql.sources.partitionOverwriteMode est static.

L'exemple suivant utilise partitionOverwriteMode:

SQL

SQL

SET spark.sql.sources.partitionOverwriteMode=dynamic;
INSERT OVERWRITE TABLE default.people10m SELECT * FROM morePeople;

Python

Python

(df.write
  .mode("overwrite")
  .option("partitionOverwriteMode", "dynamic")
  .saveAsTable("default.people10m")
)

Scala

Scala

df.write
  .mode("overwrite")
  .option("partitionOverwriteMode", "dynamic")
  .saveAsTable("default.people10m")

Gardez les contraintes et comportements suivants à l'esprit pour partitionOverwriteMode:

• Vous ne pouvez pas définir overwriteSchema sur true.
• Vous ne pouvez pas spécifier à la fois partitionOverwriteMode et replaceWhere dans la même opération DataFrameWriter.
• Si vous spécifiez une condition replaceWhere à l’aide d’une option DataFrameWriter, Delta Lake applique cette condition pour contrôler les données qui sont écrasées. Cette option est prioritaire sur la configuration partitionOverwriteMode au niveau de la session.
• Validez toujours que les données écrites ne touchent que les partitions attendues. Une seule ligne dans la mauvaise partition peut écraser involontairement toute la partition.