Dela via

Uppdatera Delta Lake-tabellschemat

  • Artikel

Med Delta Lake kan du uppdatera schemat för en tabell. Följande typer av ändringar stöds:

  • Lägga till nya kolumner (vid godtyckliga positioner)
  • Ändra ordning på befintliga kolumner
  • Byta namn på befintliga kolumner

Du kan göra dessa ändringar explicit med DDL eller implicit med DML.

Viktigt!

En uppdatering av ett Delta-tabellschema är en åtgärd som står i konflikt med alla samtidiga deltaskrivningsåtgärder.

När du uppdaterar ett Delta-tabellschema avslutas strömmar som läser från tabellen. Om du vill att strömmen ska fortsätta måste du starta om den. Rekommenderade metoder finns i Produktionsöverväganden för strukturerad direktuppspelning.

Uppdatera schemat uttryckligen för att lägga till kolumner

ALTER TABLE table_name ADD COLUMNS (col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name], ...)

Som standard är truenullability .

Om du vill lägga till en kolumn i ett kapslat fält använder du:

ALTER TABLE table_name ADD COLUMNS (col_name.nested_col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name], ...)

Om schemat till exempel innan du kör ALTER TABLE boxes ADD COLUMNS (colB.nested STRING AFTER field1) är:

- root
| - colA
| - colB
| +-field1
| +-field2

schemat efter är:

- root
| - colA
| - colB
| +-field1
| +-nested
| +-field2

Kommentar

Det går bara att lägga till kapslade kolumner för structs. Matriser och kartor stöds inte.

Uppdatera schemat uttryckligen för att ändra kolumnkommentare eller ordning

ALTER TABLE table_name ALTER [COLUMN] col_name (COMMENT col_comment | FIRST | AFTER colA_name)

Om du vill ändra en kolumn i ett kapslat fält använder du:

ALTER TABLE table_name ALTER [COLUMN] col_name.nested_col_name (COMMENT col_comment | FIRST | AFTER colA_name)

Om schemat till exempel innan du kör ALTER TABLE boxes ALTER COLUMN colB.field2 FIRST är:

- root
| - colA
| - colB
| +-field1
| +-field2

schemat efter är:

- root
| - colA
| - colB
| +-field2
| +-field1

Uppdatera schemat uttryckligen för att ersätta kolumner

ALTER TABLE table_name REPLACE COLUMNS (col_name1 col_type1 [COMMENT col_comment1], ...)

Till exempel när du kör följande DDL:

ALTER TABLE boxes REPLACE COLUMNS (colC STRING, colB STRUCT<field2:STRING, nested:STRING, field1:STRING>, colA STRING)

om schemat innan är:

- root
| - colA
| - colB
| +-field1
| +-field2

schemat efter är:

- root
| - colC
| - colB
| +-field2
| +-nested
| +-field1
| - colA

Uppdatera schemat för att byta namn på kolumner

Kommentar

Den här funktionen är tillgänglig i Databricks Runtime 10.4 LTS och senare.

Om du vill byta namn på kolumner utan att skriva om någon av kolumnernas befintliga data måste du aktivera kolumnmappning för tabellen. Se Byt namn på och släpp kolumner med Delta Lake-kolumnmappning.

Så här byter du namn på en kolumn:

ALTER TABLE table_name RENAME COLUMN old_col_name TO new_col_name

Så här byter du namn på ett kapslat fält:

ALTER TABLE table_name RENAME COLUMN col_name.old_nested_field TO new_nested_field

När du till exempel kör följande kommando:

ALTER TABLE boxes RENAME COLUMN colB.field1 TO field001

Om schemat innan är:

- root
| - colA
| - colB
| +-field1
| +-field2

Sedan är schemat efter:

- root
| - colA
| - colB
| +-field001
| +-field2

Se Byt namn på och släpp kolumner med Delta Lake-kolumnmappning.

Uppdatera schemat uttryckligen för att släppa kolumner

Kommentar

Den här funktionen är tillgänglig i Databricks Runtime 11.3 LTS och senare.

Om du vill släppa kolumner som en endast metadataåtgärd utan att skriva om några datafiler måste du aktivera kolumnmappning för tabellen. Se Byt namn på och släpp kolumner med Delta Lake-kolumnmappning.

Viktigt!

Om du tar bort en kolumn från metadata tas inte underliggande data bort för kolumnen i filer. Om du vill rensa borttagna kolumndata kan du använda REORG TABLE för att skriva om filer. Du kan sedan använda VACUUM för att fysiskt ta bort de filer som innehåller borttagna kolumndata.

Så här släpper du en kolumn:

ALTER TABLE table_name DROP COLUMN col_name

Så här släpper du flera kolumner:

ALTER TABLE table_name DROP COLUMNS (col_name_1, col_name_2)

Uppdatera schemat explicit för att ändra kolumntyp eller namn

Du kan ändra en kolumns typ eller namn eller släppa en kolumn genom att skriva om tabellen. Använd alternativet overwriteSchema för att göra detta.

I följande exempel visas hur du ändrar en kolumntyp:

(spark.read.table(...)
  .withColumn("birthDate", col("birthDate").cast("date"))
  .write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable(...)
)

I följande exempel visas hur du ändrar ett kolumnnamn:

(spark.read.table(...)
  .withColumnRenamed("dateOfBirth", "birthDate")
  .write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable(...)
)

Aktivera schemautveckling

Du kan aktivera schemautveckling genom att göra något av följande:

Databricks rekommenderar att du aktiverar schemautveckling för varje skrivåtgärd i stället för att ange en Spark-konfiguration.

När du använder alternativ eller syntax för att aktivera schemautveckling i en skrivåtgärd har detta företräde framför Spark-konfigurationen.

Kommentar

Det finns ingen schemautvecklingsklausul för INSERT INTO -instruktioner.

Aktivera schemautveckling för skrivningar för att lägga till nya kolumner

Kolumner som finns i källfrågan men som saknas i måltabellen läggs automatiskt till som en del av en skrivtransaktion när schemautvecklingen är aktiverad. Se Aktivera schemautveckling.

Skiftläge bevaras när en ny kolumn läggs till. Nya kolumner läggs till i slutet av tabellschemat. Om de ytterligare kolumnerna finns i en struct läggs de till i slutet av structen i måltabellen.

I följande exempel visas hur du använder mergeSchema alternativet med Automatisk inläsning. Se Vad är automatisk inläsare?.

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", "<path-to-schema-location>")
  .load("<path-to-source-data>")
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", "<path-to-checkpoint>")
  .trigger(availableNow=True)
  .toTable("table_name")
)

I följande exempel visas hur du använder mergeSchema alternativet med en batchskrivningsåtgärd:

(spark.read
  .table(source_table)
  .write
  .option("mergeSchema", "true")
  .mode("append")
  .saveAsTable("table_name")
)

Automatisk schemautveckling för Delta Lake-sammanslagning

Schemautveckling gör det möjligt för användare att lösa schemamatchningar mellan mål- och källtabellen i sammanslagning. Den hanterar följande två fall:

  1. En kolumn i källtabellen finns inte i måltabellen. Den nya kolumnen läggs till i målschemat och dess värden infogas eller uppdateras med hjälp av källvärdena.
  2. En kolumn i måltabellen finns inte i källtabellen. Målschemat lämnas oförändrat. värdena i den ytterligare målkolumnen lämnas antingen oförändrade (för UPDATE) eller inställda på NULL (för INSERT).

Du måste aktivera automatisk schemautveckling manuellt. Se Aktivera schemautveckling.

Kommentar

I Databricks Runtime 12.2 LTS och senare kan kolumner och structfält som finns i källtabellen anges med namn i infognings- eller uppdateringsåtgärder. I Databricks Runtime 11.3 LTS och nedan kan endast INSERT * åtgärder eller UPDATE SET * åtgärder användas för schemautveckling med sammanslagning.

I Databricks Runtime 13.3 LTS och senare kan du använda schemautveckling med structs kapslade inuti kartor, till exempel map<int, struct<a: int, b: int>>.

Schemautvecklingssyntax för sammanslagning

I Databricks Runtime 15.2 och senare kan du ange schemautveckling i en kopplingsinstruktion med SQL- eller Delta-tabell-API:er:

SQL

MERGE WITH SCHEMA EVOLUTION 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

Python

from delta.tables import *

(targetTable
  .merge(sourceDF, "source.key = target.key")
  .withSchemaEvolution()
  .whenMatchedUpdateAll()
  .whenNotMatchedInsertAll()
  .whenNotMatchedBySourceDelete()
  .execute()
)

Scala

import io.delta.tables._

targetTable
  .merge(sourceDF, "source.key = target.key")
  .withSchemaEvolution()
  .whenMatched()
  .updateAll()
  .whenNotMatched()
  .insertAll()
  .whenNotMatchedBySource()
  .delete()
  .execute()

Exempelåtgärder för sammanslagning med schemautveckling

Här är några exempel på effekterna av merge åtgärden med och utan schemautveckling.

Kolumner Fråga (i SQL) Beteende utan schemautveckling (standard) Beteende med schemautveckling
Målkolumner: key, value

Källkolumner: key, value, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED
THEN UPDATE SET *
WHEN NOT MATCHED
THEN INSERT *		 Tabellschemat förblir oförändrat. endast kolumner key, value uppdateras/infogas. Tabellschemat ändras till (key, value, new_value). Befintliga poster med matchningar uppdateras med value och new_value i källan. Nya rader infogas med schemat (key, value, new_value).
Målkolumner: key, old_value

Källkolumner: key, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED
THEN UPDATE SET *
WHEN NOT MATCHED
THEN INSERT *		 UPDATE och INSERT åtgärder utlöser ett fel eftersom målkolumnen old_value inte finns i källan. Tabellschemat ändras till (key, old_value, new_value). Befintliga poster med matchningar uppdateras med new_value i källan lämnar old_value oförändrade. Nya poster infogas med angiven key, new_valueoch NULL för old_value.
Målkolumner: key, old_value

Källkolumner: key, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED
THEN UPDATE SET new_value = s.new_value		 UPDATE genererar ett fel eftersom kolumnen new_value inte finns i måltabellen. Tabellschemat ändras till (key, old_value, new_value). Befintliga poster med matchningar uppdateras med new_value i källan lämnar old_value oförändrade och omatchade poster har NULL angetts för new_value. Se anteckning (1).
Målkolumner: key, old_value

Källkolumner: key, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN NOT MATCHED
THEN INSERT (key, new_value) VALUES (s.key, s.new_value)		 INSERT genererar ett fel eftersom kolumnen new_value inte finns i måltabellen. Tabellschemat ändras till (key, old_value, new_value). Nya poster infogas med angiven key, new_valueoch NULL för old_value. Befintliga poster har NULL angetts för new_value att lämna old_value oförändrade. Se anteckning (1).

(1) Det här beteendet är tillgängligt i Databricks Runtime 12.2 LTS och senare. Databricks Runtime 11.3 LTS och nedanstående fel i det här villkoret.

Exkludera kolumner med Delta Lake-sammanslagning

I Databricks Runtime 12.2 LTS och senare kan du använda EXCEPT satser i kopplingsvillkor för att uttryckligen undanta kolumner. Beteendet för nyckelordet EXCEPT varierar beroende på om schemautvecklingen är aktiverad eller inte.

När schemautvecklingen är inaktiverad gäller nyckelordet EXCEPT för listan med kolumner i måltabellen och tillåter exkludering av kolumner från UPDATE eller INSERT åtgärder. Exkluderade kolumner är inställda på null.

När schemautvecklingen är aktiverad gäller nyckelordet EXCEPT för listan över kolumner i källtabellen och tillåter exkludering av kolumner från schemautveckling. En ny kolumn i källan som inte finns i målet läggs inte till i målschemat om den visas i EXCEPT -satsen. Exkluderade kolumner som redan finns i målet är inställda på null.

Följande exempel visar den här syntaxen:

Kolumner Fråga (i SQL) Beteende utan schemautveckling (standard) Beteende med schemautveckling
Målkolumner: id, title, last_updated

Källkolumner: id, title, review, last_updated		 MERGE INTO target t
USING source s
ON t.id = s.id
WHEN MATCHED
THEN UPDATE SET last_updated = current_date()
WHEN NOT MATCHED
THEN INSERT * EXCEPT (last_updated)		 Matchade rader uppdateras genom att fältet anges last_updated till aktuellt datum. Nya rader infogas med värden för id och title. Det exkluderade fältet last_updated är inställt på null. Fältet review ignoreras eftersom det inte finns i målet. Matchade rader uppdateras genom att fältet anges last_updated till aktuellt datum. Schemat har utvecklats för att lägga till fältet review. Nya rader infogas med alla källfält utom last_updated som är inställda på null.
Målkolumner: id, title, last_updated

Källkolumner: id, title, review, internal_count		 MERGE INTO target t
USING source s
ON t.id = s.id
WHEN MATCHED
THEN UPDATE SET last_updated = current_date()
WHEN NOT MATCHED
THEN INSERT * EXCEPT (last_updated, internal_count)		 INSERT genererar ett fel eftersom kolumnen internal_count inte finns i måltabellen. Matchade rader uppdateras genom att fältet anges last_updated till aktuellt datum. Fältet review läggs till i måltabellen, men fältet internal_count ignoreras. Nya infogade rader har last_updated angetts till null.

NullType Hantera kolumner i schemauppdateringar

Eftersom Parquet inte stöder NullTypeNullType tas kolumnerna bort från DataFrame när de skrivs till Delta-tabeller, men lagras fortfarande i schemat. När en annan datatyp tas emot för den kolumnen sammanfogar Delta Lake schemat till den nya datatypen. Om Delta Lake tar emot en NullType för en befintlig kolumn behålls det gamla schemat och den nya kolumnen tas bort under skrivning.

NullType i strömning stöds inte. Eftersom du måste ange scheman när du använder direktuppspelning bör detta vara mycket sällsynt. NullType accepteras inte heller för komplexa typer som ArrayType och MapType.

Ersätta tabellschema

Om du skriver över data i en tabell skrivs inte schemat över som standard. När du skriver över en tabell med mode("overwrite") utan replaceWherekanske du fortfarande vill skriva över schemat för de data som skrivs. Du ersätter schemat och partitioneringen av tabellen genom att ange overwriteSchema alternativet till true:

df.write.option("overwriteSchema", "true")

Viktigt!

Du kan inte ange overwriteSchema som true när du använder dynamisk partitionsöverskrivning.