Selektives Überschreiben von Daten mit Delta Lake

Delta Lake verfügt über die folgenden unterschiedlichen Optionen für selektive Überschreibungen:

• Die Option replaceWhere ersetzt automatisch alle Datensätze, die einem bestimmten Prädikat entsprechen.
• Sie können Datenverzeichnisse mithilfe dynamischer Partitionsüberschreibungen basierend darauf ersetzen, wie Tabellen partitioniert sind.

Für die meisten Vorgänge empfiehlt Databricks die Verwendung von replaceWhere, um anzugeben, welche Daten überschrieben werden sollen.

Willkürliches selektives Überschreiben mit replaceWhere

Sie können selektiv nur die Daten überschreiben, die einem beliebigen Ausdruck entsprechen.

Ersetzen Sie z. B. Ereignisse im Januar in der Zieltabelle, die durch start_date partitioniert ist, atomar mit den Daten in replace_data.

Python

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

Scala

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

SQL

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

Dieser Beispielcode schreibt die Daten aus replace_data, überprüft, ob alle Zeilen mit dem Prädikat übereinstimmen, und führt mithilfe der overwrite-Semantik eine atomische Ersetzung durch. Wenn Werte im Vorgang außerhalb der Einschränkung liegen, tritt standardmäßig ein Fehler auf.

Sie können dieses Verhalten zu overwrite-Werten innerhalb des Prädikatbereichs und zu insert-Datensätzen ändern, die außerhalb des angegebenen Bereichs liegen. Deaktivieren Sie dazu die Überprüfung von Einschränkungen, indem Sie spark.databricks.delta.replaceWhere.constraintCheck.enabled mithilfe einer der folgenden Einstellungen auf FALSE festlegen:

Python

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

Scala

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

SQL

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

altes Verhalten

Wenn Sie das veraltete Verhalten von replaceWhere verwenden, stimmen Abfragen nur Prädikate über Partitionsspalten ab und ignorieren alle Daten, die außerhalb des Prädikatbereichs liegen. Der folgende Befehl würde den Monat Januar in der Zieltabelle, die durch date partitioniert ist, atomisch mit den Daten in df ersetzen.

Python

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

Scala

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

Um das Legacyverhalten zu verwenden, legen Sie das spark.databricks.delta.replaceWhere.dataColumns.enabled-Kennzeichen auf false:

Python

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

Scala

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

SQL

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

Dynamisches Überschreiben von Partitionen

Dynamische Partition überschreibt nur die Partitionen, bei denen der Schreibvorgang neue Daten übernimmt und andere Partitionen unverändert lässt.

Azure Databricks empfiehlt REPLACE USING, dynamische Partitionsüberschreibungen auszulösen. Dieser Modus funktioniert für alle Computetypen, einschließlich Databricks SQL Warehouses, serverloser Compute und klassischer Compute und erfordert nicht, dass Sie eine Spark-Sitzungskonfiguration festlegen. Siehe Dynamische Partitionsüberschreibungen mit REPLACE USING.

partitionOverwriteMode ist ein Legacymodus für dynamische Partitionsüberschreibungen, für die Sie die klassische Berechnung verwenden und eine Spark-Sitzungskonfiguration festlegen müssen. Es wird für Databricks SQL oder serverlose Berechnung nicht unterstützt. Siehe dynamische Partitionsüberschreibungen mit partitionOverwriteMode (legacy)

In den folgenden Abschnitten wird die Verwendung der einzelnen Modus veranschaulicht.

Dynamische Partition überschreibt mit REPLACE USING

Databricks Runtime 16.3 und höher unterstützen dynamische Partitionsüberschreibungen für Tabellen mit REPLACE USING. Sie können Daten selektiv über alle Computetypen hinweg überschreiben, ohne eine Spark-Sitzungskonfiguration festlegen zu müssen. REPLACE USING ermöglicht ein compute-unabhängiges, atomares Überschreibverhalten, das in Databricks SQL-Warehouses, serverlosem Computing und klassischem Computing funktioniert.

REPLACE USING überschreibt nur die Partitionen, die von den eingehenden Daten bestimmt sind. Alle anderen Partitionen bleiben unverändert.

REPLACE USING wird nur in SQL unterstützt. Ausführliche Informationen finden Sie in INSERT der SQL-Sprachreferenz.

In den folgenden Beispielen wird REPLACE USING verwendet.

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

Beachten Sie die folgenden Einschränkungen und Verhaltensweisen für dynamische Partitionsüberschreibungen:

• In Databricks Runtime 17.2 und höher werden partitionierte Tabellen, nicht partitionierte Tabellen und Tabellen mit flüssigem Clustering unterstützt.
• In Databricks Runtime 16.3 bis 17.1 werden nur partitionierte Tabellen unterstützt, und Sie müssen den vollständigen Satz der Partitionsspalten der Tabelle in der USING Klausel angeben.
• Überprüfen Sie immer, ob die geschriebenen Daten nur die erwarteten Partitionen berühren. Eine einzelne Zeile in der falschen Partition kann unbeabsichtigt die gesamte Partition überschreiben.

Wenn Sie eine anpassbarere Abgleichslogik benötigen, als dies REPLACE USING unterstützt wird, z. B. das Behandeln von NULL Werten als gleich, verwenden Sie stattdessen die komplementierende REPLACE ON Logik. Weitere Informationen finden Sie unter INSERT.

Dynamische Partition überschreibt mit partitionOverwriteMode (veraltet)

Databricks Runtime 11.3 LTS und höher unterstützt das dynamische Überschreiben von Partitionen bei partitionierten Tabellen im Überschreibmodus: entweder INSERT OVERWRITE in SQL oder mittels eines DataFrame-Schreibzugriffs mit df.write.mode("overwrite"). Diese Art der Überschreibung ist nur für klassische Rechenleistung verfügbar und nicht für Databricks SQL Warehouses oder serverlose Rechenleistung.

Wenn Sie den Modus für dynamische Partitionsüberschreibung verwenden möchten, legen Sie die Spark-Sitzungskonfiguration spark.sql.sources.partitionOverwriteMode auf dynamic. Sie können alternativ die DataFrameWriter Option partitionOverwriteMode auf dynamic setzen. Wenn vorhanden, setzt die abfragespezifische Option den in der Sitzungskonfiguration definierten Modus außer Kraft. Der Standardwert für spark.sql.sources.partitionOverwriteMode ist static.

Im folgenden Beispiel wird partitionOverwriteMode verwendet:

SQL

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

Python

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

Scala

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

Beachten Sie die folgenden Einschränkungen und Verhaltensweisen für partitionOverwriteMode:

• Sie können overwriteSchema nicht auf true setzen.
• Sie können nicht sowohl partitionOverwriteMode als auch replaceWhere im selben DataFrameWriter-Vorgang angeben.
• Wenn Sie eine replaceWhere Bedingung mithilfe einer DataFrameWriter Option angeben, wendet Delta Lake diese Bedingung an, um zu steuern, welche Daten überschrieben werden. Diese Option hat Vorrang vor der partitionOverwriteMode Konfiguration auf Sitzungsebene.
• Überprüfen Sie immer, ob die geschriebenen Daten nur die erwarteten Partitionen berühren. Eine einzelne Zeile in der falschen Partition kann unbeabsichtigt die gesamte Partition überschreiben.