Breddning av typer

I Delta Lake-tabeller i Databricks Runtime 15.4 LTS och senare kan du ändra kolumndatatyper till en bredare typ utan att skriva om datafiler.

Alla hanterade Unity Catalog-tabeller använder Delta Lake som standard. Se Hanterade tabeller i Unity Catalog för Delta Lake och Apache Iceberg.

Typändringar som stöds

Du kan bredda typerna enligt följande regler:

Typändringar stöds för kolumner och fält på den översta nivån som är kapslade i strukturer, kartor och matriser.

Decimalbeteende

Spark trunkerar bråkdelen av ett värde som standard när en åtgärd höjer upp en heltalstyp till en decimal eller double och en nedströmsinmatning skriver tillbaka värdet till en heltalskolumn. Mer information om beteendet för tilldelningsprinciper finns i Butikstilldelning.

När du ändrar en numerisk typ till decimalmåste den totala precisionen vara lika med eller större än startprecisionen. Om du också ökar skalan måste den totala precisionen öka med motsvarande mängd.

Det minsta målet för byte, shortoch int typer är decimal(10,0). Det minsta målet för long är decimal(20,0).

Om du vill lägga till två decimaler i ett fält med decimal(10,1)är det lägsta målet decimal(12,3).

Aktivera typbreddning

Du kan aktivera typbreddning i en befintlig tabell genom att ange tabellegenskapen delta.enableTypeWidening till true:

  ALTER TABLE <table_name> SET TBLPROPERTIES ('delta.enableTypeWidening' = 'true')

Du kan också aktivera typbreddning när tabellen skapas:

  CREATE TABLE T(c1 INT) TBLPROPERTIES('delta.enableTypeWidening' = 'true')

Tillämpa en typändring manuellt

ALTER COLUMN Använd kommandot för att ändra typer manuellt:

ALTER TABLE <table_name> ALTER COLUMN <col_name> TYPE <new_type>

Den här åtgärden uppdaterar tabellschemat utan att skriva om de underliggande datafilerna. Mer information finns i ALTER TABLE.

Bredda typer med automatisk schemautveckling

Använd schemautveckling med typbreddning för att uppdatera datatyper i måltabeller för att matcha typen av inkommande data.

Om du vill använda schemautveckling för att bredda datatypen för en kolumn under inmatningen måste du uppfylla följande villkor:

• Skrivkommandot körs med automatisk schemautveckling aktiverad.
• Måltabellen har typbreddning aktiverad.
• Källkolumntypen är bredare än målkolumntypen.
• Typbreddning stödjer typändringen.

Typmatchningar som inte uppfyller alla dessa villkor följer normala regler för schematillämpning. Se Genomdrivande av schema.

Example

Följande exempel visar hur typbreddning fungerar med schemautveckling.

Python

Skapa en måltabell med en INT kolumn och en källtabell med en BIGINT kolumn:

spark.sql("CREATE TABLE target_table (id INT, data STRING) TBLPROPERTIES ('delta.enableTypeWidening' = 'true')")
spark.sql("CREATE TABLE source_table (id BIGINT, data STRING)")

Använd saveAsTable() med schemautveckling för att automatiskt bredda INT kolumnen till BIGINT under ett tillägg:

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

Använd MERGE INTO med schemautveckling:

from delta.tables import DeltaTable

source_df = spark.table("source_table")
target_table = DeltaTable.forName(spark, "target_table")

(target_table.alias("target")
  .merge(source_df.alias("source"), "target.id = source.id")
  .withSchemaEvolution()
  .whenMatchedUpdateAll()
  .whenNotMatchedInsertAll()
  .execute()
)

Scala

Skapa en måltabell med en INT kolumn och en källtabell med en BIGINT kolumn:

spark.sql("CREATE TABLE target_table (id INT, data STRING) TBLPROPERTIES ('delta.enableTypeWidening' = 'true')")
spark.sql("CREATE TABLE source_table (id BIGINT, data STRING)")

Använd saveAsTable() med schemautveckling för att automatiskt bredda INT kolumnen till BIGINT under ett tillägg:

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

Använd MERGE INTO med schemautveckling:

import io.delta.tables.DeltaTable

val sourceDf = spark.table("source_table")
val targetTable = DeltaTable.forName(spark, "target_table")

targetTable.alias("target")
  .merge(sourceDf.alias("source"), "target.id = source.id")
  .withSchemaEvolution()
  .whenMatched().updateAll()
  .whenNotMatched().insertAll()
  .execute()

SQL

Skapa en måltabell med en INT kolumn och en källtabell med en BIGINT kolumn:

CREATE TABLE target_table (id INT, data STRING) TBLPROPERTIES ('delta.enableTypeWidening' = 'true');
CREATE TABLE source_table (id BIGINT, data STRING);

Använd INSERT INTO med schemautveckling för att automatiskt bredda INT kolumnen till BIGINT under ett tillägg:

INSERT WITH SCHEMA EVOLUTION INTO target_table SELECT * FROM source_table;

Använd MERGE INTO med schemautveckling:

MERGE WITH SCHEMA EVOLUTION INTO target_table
USING source_table
ON target_table.id = source_table.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

Automatisk Laddare

Auto Loader stöder typbreddning med automatisk schemautveckling. När du använder Auto Loader för att mata in data i en Delta Lake-tabell med typbreddning och schemautveckling aktiverad, breddas kolumntyper automatiskt så att de matchar inkommande data.

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

Se Automatisk typbreddning med Auto Loader. Måltabellen måste också ha aktiverat typbreddning. Se Aktivera typbreddning.

Inaktivera funktionen för typbreddning av tabell

Du kan förhindra att oavsiktlig typbreddning på aktiverade tabeller aktiveras genom att ställa in egenskapen på false:

  ALTER TABLE <table_name> SET TBLPROPERTIES ('delta.enableTypeWidening' = 'false')

Den här inställningen förhindrar framtida typändringar i tabellen, men tar inte bort funktionen för att bredda tabellen eller ångra tidigare typändringar.

Om du behöver ta bort tabellfunktionerna för typbreddning helt kan du använda DROP FEATURE kommandot som du ser i följande exempel:

 ALTER TABLE <table-name> DROP FEATURE 'typeWidening' [TRUNCATE HISTORY]

När du släpper typbreddning skriver Databricks om alla datafiler som inte överensstämmer med det aktuella tabellschemat. Se Ta bort en Delta Lake-tabellfunktion och nedgradera tabellprotokollet.

Strömma från en Delta Lake-tabell

Stöd för typbreddning i Strukturerad direktuppspelning finns i Databricks Runtime 16.4 LTS och senare.

När du strömmar från en Delta Lake-tabell med typbreddning aktiverat kan du konfigurera automatisk typbreddning för strömmande frågor genom att aktivera schemautveckling med mergeSchema alternativet på måltabellen. Måltabellen måste ha aktiverat typbreddning. Se Aktivera typbreddning.

Python

(spark.readStream
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", "/path/to/checkpointLocation")
  .option("mergeSchema", "true")
  .toTable("output_table")
)

Scala

spark.readStream
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", "/path/to/checkpointLocation")
  .option("mergeSchema", "true")
  .toTable("output_table")

När mergeSchema är aktiverat och måltabellen har typbreddning aktiverat:

• Typändringar tillämpas automatiskt på den underordnade tabellen utan manuella åtgärder.
• Nya kolumner läggs automatiskt till i det underordnade tabellschemat.

Utan mergeSchema aktiverad hanteras värden enligt konfigurationen spark.sql.storeAssignmentPolicy , som som standard nedarbetar värden för att matcha målkolumntypen. Mer information om tilldelningspolicyers beteende finns i Store assignment.

Hantera typändringar i en dataström

När du strömmar från en Delta Lake-tabell kan du ange en plats för schemaspårning för att spåra icke-additiva schemaändringar, inklusive typändringar. Att tillhandahålla en plats för schemaspårning krävs i Databricks Runtime 18.0 och senare, och det är valfritt i Databricks Runtime 18.1 och senare.

Du kan inte ange en schemaTrackingLocation med SQL. Se Funktioner som inte stöds.

schemaTrackingLocation måste anges till en plats inom samma sökväg som din kontrollpunkt för strömning. Ett exempel:

Python

checkpoint_path = "/path/to/checkpointLocation"

(spark.readStream
  .option("schemaTrackingLocation", checkpoint_path)
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("output_table")
)

Scala

val checkpointPath = "/path/to/checkpointLocation"

spark.readStream
  .option("schemaTrackingLocation", checkpointPath)
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpointPath)
  .toTable("output_table")

När du har angett en plats för schemaspårning utvecklar strömmen sitt spårade schema när den identifierar en typändring och stoppar sedan. Då måste du hantera typändringen, till exempel genom att aktivera typbreddning på måltabellen längre ned i flödet eller genom att uppdatera streamingfrågan.

Om du vill återuppta bearbetningen anger du Spark-konfigurationen spark.databricks.delta.streaming.allowSourceColumnTypeChange eller läsaralternativet DataFrameallowSourceColumnTypeChange, som i följande exempel:

Python

checkpoint_path = "/path/to/checkpointLocation"

(spark.readStream
  .option("schemaTrackingLocation", checkpoint_path)
  .option("allowSourceColumnTypeChange", "<delta_source_table_version>")
  # alternatively to allow all future type changes for this stream:
  # .option("allowSourceColumnTypeChange", "always")
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("output_table")
)

Scala

val checkpointPath = "/path/to/checkpointLocation"

spark.readStream
  .option("schemaTrackingLocation", checkpointPath)
  .option("allowSourceColumnTypeChange", "<delta_source_table_version>")
  // alternatively to allow all future type changes for this stream:
  // .option("allowSourceColumnTypeChange", "always")
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpointPath)
  .toTable("output_table")

SQL

  -- To unblock for this particular stream just for this series of schema change(s):
  SET spark.databricks.delta.streaming.allowSourceColumnTypeChange.ckpt_<checkpoint_id> = "<delta_source_table_version>"
  -- To unblock for this particular stream:
  SET spark.databricks.delta.streaming.allowSourceColumnTypeChange = "<delta_source_table_version>"
  -- To unblock for all streams:
  SET spark.databricks.delta.streaming.allowSourceColumnTypeChange = "always"

När dataströmmen stoppas visas kontrollpunkts-ID:t <checkpoint_id> och versionen för Delta Lake-källtabellen <delta_source_table_version> i ett felmeddelande.

En fullständig lista över alternativ för strömning av Delta Lake finns i Delta Lake.

Lakeflow-pipelines

Du kan aktivera typbreddning för Lakeflow-pipelines på pipelinenivå eller för enskilda tabeller. Med typbreddning kan kolumntyperna utökas automatiskt under pipelinekörningen utan att det krävs en fullständig uppdatering av strömmande tabeller. Typändringar i materialiserade vyer utlöser alltid en fullständig omkompensering, och när en typändring tillämpas på en källtabell kräver materialiserade vyer som är beroende av den tabellen en fullständig omkompensering för att återspegla de nya typerna.

Aktivera typbreddning för en hel pipeline

Om du vill aktivera typbreddning för alla tabeller i en pipeline anger du pipelinekonfigurationen pipelines.enableTypeWidening:

JSON

{
  "configuration": {
    "pipelines.enableTypeWidening": "true"
  }
}

YAML

configuration:
  pipelines.enableTypeWidening: 'true'

Aktivera typbreddning för specifika tabeller

Du kan också aktivera typbreddning för enskilda tabeller genom att ange tabellegenskapen delta.enableTypeWidening:

Python

import dlt

@dlt.table(
  table_properties={"delta.enableTypeWidening": "true"}
)
def my_table():
  return spark.readStream.table("source_table")

SQL

CREATE OR REFRESH STREAMING TABLE my_table
TBLPROPERTIES ('delta.enableTypeWidening' = 'true')
AS SELECT * FROM source_table

Kompatibilitet med underordnade läsare

Tabeller med typbreddning aktiverat kan bara läsas i Databricks Runtime 15.4 LTS och senare. Om du vill att en tabell med typbreddning aktiverad i pipelinen ska kunna läsas av läsare på Databricks Runtime 14.3 och nedan måste du antingen:

• Inaktivera typbreddning genom att ta bort egenskapen delta.enableTypeWidening/pipelines.enableTypeWidening eller ställa in den på false och utlösa en fullständig uppdatering av tabellen.
• Aktivera kompatibilitetsläge i tabellen.

OpenSharing

Delning av en Delta Lake-tabell med typbreddning aktiverat stöds i Databricks-till-Databricks OpenSharing. Providern och mottagaren måste vara på Databricks Runtime 16.1 eller senare.

Om du vill läsa ändringsdataflöde från en Delta Lake-tabell med typbreddning aktiverat med OpenSharing måste du ange svarsformatet till delta:

spark.read
  .format("deltaSharing")
  .option("responseFormat", "delta")
  .option("readChangeFeed", "true")
  .option("startingVersion", "<start version>")
  .option("endingVersion", "<end version>")
  .load("<table>")

Det går inte att läsa ändringsdataflöde mellan typändringar. Du måste i stället dela upp åtgärden i två separata läsningar, en som slutar vid tabellversionen som innehåller typändringen och den andra med början i den version som innehåller typändringen.

Limitations

Apache Iceberg-kompatibilitet

Apache Iceberg stöder inte alla typändringar som omfattas av typbreddning. Se Utveckling av isbergsschema.

Ändringar av typen som inte stöds omfattar följande:

• byte, short, int, long till decimal eller double
• ökning av decimalskala
• date till timestampNTZ

När du aktiverar Iceberg-läsningar i en Delta Lake-tabell resulterar ett fel i en av de föregående typändringarna. Läs Delta Lake-tabeller med Iceberg-klienter.

Om du tillämpar någon av dessa typändringar som inte stöds i en Delta Lake-tabell har du två alternativ:

• Återskapa Iceberg-metadata: Använd följande kommando för att återskapa Iceberg-metadata utan funktionen för att bredda tabellen:

  ALTER TABLE <table-name> SET TBLPROPERTIES ('delta.universalFormat.config.icebergCompatVersion' = '<version>')
  

  Detta gör att du kan behålla kompatibiliteten med Iceberg-läsningar efter att ha gjort ändringar av inkompatibla typer.

• Släpp funktionen för typbreddningstabell: Se Inaktivera funktionen för att bredda tabellen.

Typberoende funktioner

Vissa SQL-funktioner returnerar resultat som är beroende av indatatypen. Funktionen returnerar till exempel hash olika hash-värden för samma logiska värde om argumenttypen är annorlunda: hash(1::INT) returnerar ett annat resultat än hash(1::BIGINT).

Andra typberoende funktioner är: xxhash64, bit_get, bit_reverse, typeof.

För stabila resultat i frågor som använder dessa funktioner måste du uttryckligen omvandla värden till önskad typ:

Python

spark.read.table("table_name") \
  .selectExpr("hash(CAST(column_name AS BIGINT))")

Scala

spark.read.table("main.johan_lasperas.dlt_type_widening_bronze2")
  .selectExpr("hash(CAST(a AS BIGINT))")

SQL

-- Use explicit casting for stable hash values
SELECT hash(CAST(column_name AS BIGINT)) FROM table_name

Funktioner som inte stöds

• Du kan inte ange en plats för schemaspårning med SQL när du strömmar från en Delta Lake-tabell med en typändring.
• Du kan inte dela en tabell med typbreddning aktiverat för användare som inte är Databricks-användare med OpenSharing.