Sdílet prostřednictvím

Selektivní přepsání dat pomocí Delta Lake

  • Článek

Azure Databricks využívá funkci Delta Lake k podpoře dvou různých možností selektivního přepsání:

  • Možnost replaceWhere atomicky nahradí všechny záznamy, které odpovídají danému predikátu.
  • Datové složky můžete nahradit na základě způsobu dělení tabulek pomocí dynamického přepisování oddílů.

Pro většinu operací doporučuje replaceWhere Databricks určit, která data se mají přepsat.

Důležité

Pokud byla data omylem přepsána, můžete změnu vrátit zpět pomocí obnovení.

Libovolná selektivní přepsání pomocí replaceWhere

Můžete selektivně přepsat pouze data, která odpovídají libovolnému výrazu.

Poznámka:

SQL vyžaduje Databricks Runtime 12.2 LTS nebo vyšší.

Následující příkaz atomicky nahradí události za leden v cílové tabulce, která je rozdělena na základě start_date, daty z 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

Tento ukázkový kód zapíše data do replace_data, ověří, že všechny řádky odpovídají predikátu a provádí atomické nahrazení pomocí overwrite sémantiky. Pokud některé hodnoty v operaci spadají mimo omezení, tato operace ve výchozím nastavení selže s chybou.

Můžete změnit toto chování tak, aby se overwrite hodnoty přizpůsobily v rámci rozsahu daného predikátu a aby insert záznamy spadaly mimo určený rozsah. Pokud to chcete udělat, zakažte kontrolu omezení nastavením spark.databricks.delta.replaceWhere.constraintCheck.enabled na false pomocí jednoho z následujících nastavení:

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

Chování starší verze

Starší výchozí chování znamenalo, že replaceWhere přepsalo data odpovídající predikátu pouze u sloupců oddílů. V tomto starším modelu by následující příkaz atomicky nahradil měsíc leden v cílové tabulce, který je rozdělený podle date, s daty v 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")

Pokud se chcete vrátit ke starému chování, můžete příznak zakázat spark.databricks.delta.replaceWhere.dataColumns.enabled :

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

dynamické přepsání oddílu

Důležité

Tato funkce je ve verzi Public Preview.

Databricks Runtime 11.3 LTS a vyšší podporuje dynamický režim přepsání pro dělené tabulky. U tabulek s více oddíly podporují Databricks Runtime 11.3 LTS a níže pouze dynamické přepsání oddílů, pokud jsou všechny sloupce oddílů stejného datového typu.

V režimu přepisu dynamického oddílu operace přepíšou všechna existující data v každém logickém oddílu, do kterých se zapisují nová data. Všechny existující logické oddíly, pro které zápis neobsahuje data, zůstanou beze změny. Tento režim platí pouze v případě, že se data zapisují v režimu přepsání: buď INSERT OVERWRITE v SQL, nebo při zápisu do datového rámce s df.write.mode("overwrite").

Pro nastavení režimu přepsání dynamického oddílu nakonfigurujte relaci Spark z spark.sql.sources.partitionOverwriteMode na dynamic. Můžete to také povolit nastavením DataFrameWriter možnosti partitionOverwriteModedynamic. Pokud existuje, možnost specifická pro dotaz přepíše režim definovaný v konfiguraci relace. Výchozí hodnota je partitionOverwriteModestatic.

Důležité

Ověřte, že data zapsaná při přepsání dynamických oddílů zasáhnou pouze očekávané oddíly. Jeden řádek v nesprávné partition může vést k neúmyslnému přepsání celé partition.

Následující příklad ukazuje použití dynamického přepsání oddílů:

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

Poznámka:

  • Dynamické přepsání oddílů koliduje s možností replaceWhere pro particionované tabulky.
    • Pokud je v konfiguraci relace Sparku povolená možnost přepsání dynamického oddílu a replaceWhere je k dispozici jako možnost DataFrameWriter, pak Delta Lake přepíše data podle výrazu replaceWhere (možnosti specifické pro dotazy přepisují konfigurace relací).
    • Pokud jsou u možností DataFrameWriter povoleno přepisování dynamického oddílu i replaceWhere, zobrazí se chyba.
  • Při použití dynamického přepsání oddílu nelze zadat overwriteSchema jako true.