Gegevens selectief overschrijven met Delta Lake

Azure Databricks maakt gebruik van Delta Lake-functionaliteit ter ondersteuning van twee verschillende opties voor selectieve overschrijvingen:

• De replaceWhere optie vervangt alle records die overeenkomen met een bepaald predicaat.
• U kunt mappen met gegevens vervangen op basis van hoe tabellen worden gepartitioneerd met behulp van dynamische partitieoverschrijven.

Voor de meeste bewerkingen raadt replaceWhere Databricks aan om op te geven welke gegevens moeten worden overschreven.

Willekeurig selectief overschrijven met replaceWhere

U kunt selectief alleen de gegevens overschrijven die overeenkomen met een willekeurige expressie.

Met de volgende opdracht worden gebeurtenissen in januari in de doeltabel, die wordt gepartitioneerd door start_date, vervangen door de gegevens in replace_data:

Python

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

Scala

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

SQL

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

Deze voorbeeldcode schrijft de gegevens uit replace_data, valideert dat alle rijen overeenkomen met het predicaat en voert een atomische vervanging uit met behulp van overwrite semantiek. Als waarden in de bewerking buiten de beperking vallen, mislukt deze bewerking standaard met een fout.

U kunt dit gedrag wijzigen in overwrite waarden binnen het predicaatbereik en insert records die buiten het opgegeven bereik vallen. Schakel hiervoor de beperkingscontrole uit door spark.databricks.delta.replaceWhere.constraintCheck.enabled in te stellen op false met behulp van een van de volgende instellingen:

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

Verouderd gedrag

Het verouderde standaardgedrag zorgde ervoor dat replaceWhere gegevens overschreef die alleen overeenkomen met een predicaat over partitiekolommen. Met dit verouderde model zou de volgende opdracht de maand januari in de doeltabel, die wordt gepartitioneerd door date, vervangen door de gegevens in df:

Python

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

Scala

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

Als u terug wilt vallen op het oude gedrag, kunt u de spark.databricks.delta.replaceWhere.dataColumns.enabled vlag uitschakelen:

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

Overschrijvingen van dynamische partities

Databricks Runtime 11.3 LTS en hoger ondersteunt dynamische partitieoverschrijfmodus voor gepartitioneerde tabellen. Voor tabellen met meerdere partities ondersteunt Databricks Runtime 11.3 LTS en hieronder alleen dynamische partitieoverschrijven als alle partitiekolommen van hetzelfde gegevenstype zijn.

Wanneer de bewerkingen in de dynamische partitie-overschrijvingsmodus zijn, overschrijven ze alle bestaande gegevens in elke logische partitie waarvoor de schrijfbewerking nieuwe gegevens doorvoert. Bestaande logische partities waarvoor de schrijfbewerking geen gegevens bevat, blijven ongewijzigd. Deze modus is alleen van toepassing wanneer gegevens worden geschreven in de overschrijfmodus: INSERT OVERWRITE in SQL of een DataFrame-schrijfbewerking met df.write.mode("overwrite").

Configureer de modus voor dynamisch overschrijven van partities door de configuratie van de Spark-sessie in te stellen spark.sql.sources.partitionOverwriteMode op dynamic. U kunt dit ook inschakelen door de DataFrameWriter optie partitionOverwriteMode in te stellen op dynamic. Indien aanwezig, overschrijft de queryspecifieke optie de modus die is gedefinieerd in de sessieconfiguratie. De standaardwaarde partitionOverwriteMode is static.

In het volgende voorbeeld ziet u hoe dynamische partities worden overschreven:

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")