Type-uitbreiding

Beschikbaar voor Delta Lake-tabellen in Databricks Runtime 15.4 LTS en hoger. Met het type widening kunt u kolomgegevenstypen wijzigen in een breder type zonder gegevensbestanden opnieuw te schrijven.

Alle beheerde tabellen van Unity Catalog maken standaard gebruik van Delta Lake. Zie beheerde tabellen van Unity Catalog voor Delta Lake en Apache Iceberg.

Opmerking

Door het inschakelen van type widening worden de lezer- en schrijfprotocollen bijgewerkt. Dit kan van invloed zijn op de compatibiliteit met externe Delta Lake-clients. Bekijk de compatibiliteit en protocollen van Delta Lake-functies.

Tabellen waarvoor type widening is ingeschakeld, kunnen alleen worden gelezen door Databricks Runtime 15.4 LTS en hoger.

Ondersteunde typewijzigingen

U kunt typen breder maken op basis van de volgende regels:

Bronsoort Ondersteunde uitgebreidere typen
BYTE SHORT, INT, BIGINT, DECIMAL, , DOUBLE
SHORT INT,BIGINT,DECIMAL,DOUBLE
INT BIGINT, DECIMAL, DOUBLE
BIGINT DECIMAL
FLOAT DOUBLE
DECIMAL DECIMAL met meer precisie en schaal
DATE TIMESTAMP_NTZ
VOID Alle soorten

Typewijzigingen worden ondersteund voor kolommen en velden op het hoogste niveau die zijn genest in structs, kaarten en matrices.

Opmerking

VOID voor elk type hoeft type widening niet in de tabel te worden ingeschakeld. Elke bewerking waarmee het type kolom VOID wordt bijgewerkt, slaagt zonder extra configuratie. VOID type verbreding is beschikbaar in Databricks Runtime 18.2 en hoger.

Decimaal gedrag

Spark kapt het breukgedeelte van een waarde standaard af wanneer een bewerking een geheel getal naar een decimal of double promoot en een downstream opname de waarde terugschrijft naar een kolom met gehele getallen. Zie Store assignment voor meer informatie over het toewijzingsbeleid.

Wanneer u een numeriek type wijzigt in decimal, moet de totale precisie gelijk zijn aan of groter zijn dan de beginprecisie. Als u ook de schaal verhoogt, moet de totale precisie met een corresponderend bedrag toenemen.

Het minimumdoel voor byte, shorten int typen is decimal(10,0). Het minimumdoel voor long is decimal(20,0).

Als u twee decimalen wilt toevoegen aan een veld met decimal(10,1), is het minimumdoel decimal(12,3).

Typebreiding inschakelen

Opmerking

Door het inschakelen van type widening worden de lezer- en schrijfprotocollen bijgewerkt. Dit kan van invloed zijn op de compatibiliteit met externe Delta Lake-clients. Bekijk de compatibiliteit en protocollen van Delta Lake-functies.

U kunt type-uitbreiding inschakelen voor een bestaande tabel door de delta.enableTypeWidening tabeleigenschap in te stellen op true:

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

U kunt ook typebreiding inschakelen tijdens het maken van een tabel:

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

Handmatig een typewijziging toepassen

Gebruik de ALTER COLUMN opdracht om typen handmatig te wijzigen:

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

Met deze bewerking wordt het tabelschema bijgewerkt zonder de onderliggende gegevensbestanden opnieuw te schrijven. Zie ALTER TABLE voor meer informatie.

Breid typen uit met automatische schema-evolutie

Gebruik schemaontwikkeling met typebreiding om gegevenstypen in doeltabellen bij te werken zodat deze overeenkomen met het type binnenkomende gegevens.

Opmerking

Zonder dat type-verbreding is ingeschakeld, probeert schema-evolutie altijd gegevens te downcasten om overeen te komen met kolomtypen in de doel-tabel. Als u gegevenstypen niet automatisch wilt breder maken in uw doeltabellen, moet u typebreedheid uitschakelen voordat u workloads uitvoert waarvoor schemaontwikkeling is ingeschakeld.

Als u schemaontwikkeling wilt gebruiken om het gegevenstype van een kolom tijdens opname te verbreed, moet u aan de volgende voorwaarden voldoen:

  • De schrijfopdracht wordt uitgevoerd met automatische schemaontwikkeling ingeschakeld.
  • De doeltabel heeft 'type widening' (datatype-uitbreiding) ingeschakeld.
  • Het bronkolomtype is breder dan het doelkolomtype.
  • Type-uitbreiding ondersteunt de verandering van het type.

Typemismatches die niet aan al deze voorwaarden voldoen, volgen de normale schema-afdwingingsregels. Zie Schemahandhaving.

Example

In de volgende voorbeelden ziet u hoe type widening werkt met schemaontwikkeling.

Python

Maak een doeltabel met een INT kolom en een brontabel met een BIGINT kolom:

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

Gebruik saveAsTable() met schema-evolutie om de kolom INT automatisch te verbreden naar BIGINT tijdens een append-bewerking:

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

MERGE INTO gebruiken met schema-evolutie:

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

Maak een doeltabel met een INT kolom en een brontabel met een BIGINT kolom:

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

Gebruik saveAsTable() met schema-evolutie om de kolom INT automatisch te verbreden naar BIGINT tijdens een append-bewerking:

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

MERGE INTO gebruiken met schema-evolutie:

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

Maak een doeltabel met een INT kolom en een brontabel met een BIGINT kolom:

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

Gebruik INSERT INTO met schema-evolutie om de kolom INT automatisch te verbreden naar BIGINT tijdens een append-bewerking:

INSERT WITH SCHEMA EVOLUTION INTO target_table SELECT * FROM source_table;

MERGE INTO gebruiken met schema-evolutie:

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 *;

Automatische lader

Important

Ondersteuning voor typeverruiming in Auto Loader bevindt zich in Public Preview.

Auto Loader ondersteunt type-uitbreiding met automatische schema-evolutie. Wanneer u Automatisch laden gebruikt om gegevens op te nemen in een Delta Lake-tabel waarvoor typebreedheid en schemaontwikkeling is ingeschakeld, worden kolomtypen automatisch uitgebreid zodat deze overeenkomen met de binnenkomende gegevens.

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

Zie Automatische typeverbreding met Auto Loader. Daarnaast moet voor de doeltabel "type widening" zijn ingeschakeld. Zie Widening van type inschakelen.

De functie voor het uitbreiden van de tabeltype uitschakelen

U kunt onbedoelde typebreiding voor ingeschakelde tabellen voorkomen door de eigenschap in te stellen op false:

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

Met deze instelling voorkomt u toekomstige wijzigingen in het type in de tabel, maar verwijdert u de functie voor het breder maken van de tabel niet of verwijdert u eerdere typewijzigingen.

Als u de tabelfuncties voor het breder maken van het type volledig wilt verwijderen, kunt u de DROP FEATURE opdracht gebruiken, zoals wordt weergegeven in het volgende voorbeeld:

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

Opmerking

Voor tabellen die typeverruiming hebben ingeschakeld met Databricks Runtime 15.4 LTS, moet u in plaats daarvan de functie typeWidening-preview uitschakelen.

Wanneer u het verbreden van het type stopt, herschrijft Databricks alle gegevensbestanden die niet voldoen aan het huidige tabelschema. Zie Een Delta Lake-tabelfunctie verwijderen en het tabelprotocol degraderen.

Streamen vanuit een Delta Lake-tabel

Ondersteuning voor type widening in Structured Streaming is beschikbaar in Databricks Runtime 16.4 LTS en hoger.

Wanneer u streamt vanuit een Delta Lake-tabel waarvoor type widening is ingeschakeld, kunt u automatische typebreiding configureren voor streamingquery's door schemaontwikkeling in te schakelen met de mergeSchema optie voor de doeltabel. Voor de doeltabel moet typeuitbreiding ingeschakeld zijn. Zie Widening van type inschakelen.

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

Wanneer mergeSchema is ingeschakeld en de doeltabel ingeschakelde type-verbreding heeft:

  • Typewijzigingen worden automatisch toegepast op de downstreamtabel zonder handmatige tussenkomst.
  • Nieuwe kolommen worden automatisch toegevoegd aan het downstreamtabelschema.

Zonder dat mergeSchema is ingeschakeld, worden waarden verwerkt volgens de spark.sql.storeAssignmentPolicy-configuratie, die standaard als volgt omlaaggaat om te voldoen aan het doelkolomtype. Zie Store toewijzing voor meer informatie over het gedrag van het toewijzingsbeleid.

Typewijzigingen in een stream verwerken

Wanneer u vanuit een Delta Lake-tabel streamt, kunt u een locatie voor schematracking opgeven om niet-additieve schemawijzigingen bij te houden, inclusief typewijzigingen. Het instellen van een locatie voor schematracking is vereist in Databricks Runtime 18.0 en lager en is optioneel in Databricks Runtime 18.1 en hoger.

U kunt geen schemaTrackingLocation instellen met SQL. Zie Niet-ondersteunde functies.

schemaTrackingLocation moet worden ingesteld op een locatie binnen hetzelfde pad als uw streamingcontrolepunt. Voorbeeld:

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

Nadat u een locatie voor schematracking hebt ingesteld, ontwikkelt de stroom het bijgehouden schema wanneer een typewijziging wordt gedetecteerd en vervolgens wordt gestopt. Op dat moment moet u de wijziging van het type afhandelen, zoals het inschakelen van het breder maken van typen in de downstreamtabel of het bijwerken van de streamingquery.

Als u de verwerking wilt hervatten, stelt u de Spark-configuratie spark.databricks.delta.streaming.allowSourceColumnTypeChange of de DataFrame lezeroptie allowSourceColumnTypeChangein, zoals in het volgende voorbeeld:

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"

Wanneer de stream stopt, wordt in een foutbericht de controlepunt-id <checkpoint_id> en de versie van <delta_source_table_version>de Delta Lake-brontabel weergegeven.

Zie Delta Lake voor een volledige lijst met opties voor het streamen van Delta Lake.

Lakeflow-pijplijnen

U kunt typeverbreding voor Lakeflow-pijplijnen inschakelen op pijplijnniveau of voor afzonderlijke tabellen. Met typebreiding kunnen kolomtypen automatisch worden uitgebreid tijdens de uitvoering van de pijplijn zonder dat er een volledige vernieuwing van streamingtabellen nodig is. Typewijzigingen in gerealiseerde weergaven activeren altijd een volledige hercomputing en wanneer een typewijziging wordt toegepast op een brontabel, moeten gerealiseerde weergaven die afhankelijk zijn van die tabel een volledige hercomputing vereisen om de nieuwe typen weer te geven.

Typebreiding inschakelen voor een hele pijplijn

Als u typebreiding wilt inschakelen voor alle tabellen in een pijplijn, stelt u de pijplijnconfiguratie pipelines.enableTypeWideningin:

JSON

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

YAML

configuration:
  pipelines.enableTypeWidening: 'true'

Typebreiding inschakelen voor specifieke tabellen

U kunt ook typebreiding voor afzonderlijke tabellen inschakelen door de tabeleigenschap delta.enableTypeWideningin te stellen:

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

Compatibiliteit met downstreamlezers

Tabellen waarvoor type widening is ingeschakeld, kunnen alleen worden gelezen in Databricks Runtime 15.4 LTS en hoger. Als u wilt dat een tabel met type-verbreding ingeschakeld in uw pijplijn kan worden gelezen door lezers van Databricks Runtime 14.3 en lager, moet u een van de volgende opties gebruiken:

  • Schakel typeverbreiding uit door de eigenschap delta.enableTypeWidening/pipelines.enableTypeWidening te verwijderen of in te stellen op false en een volledige vernieuwing van de tabel te activeren.
  • Schakel de compatibiliteitsmodus in uw tabel in.

OpenSharing

Opmerking

Ondersteuning voor Type Widening in OpenSharing is beschikbaar in Databricks Runtime 16.1 en hoger.

Het delen van een Delta Lake-tabel waarvoor type widening is ingeschakeld, wordt ondersteund in Databricks-to-Databricks OpenSharing. De provider en ontvanger moeten zich in Databricks Runtime 16.1 of hoger bevindt.

Als u de wijzigingsgegevensfeed wilt lezen uit een Delta Lake-tabel waarbij typeverbreding is ingeschakeld via OpenSharing, moet u de antwoordindeling instellen op delta:

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

Het lezen van wijzigingenfeed voor verschillende typen wijzigingen wordt niet ondersteund. In plaats daarvan moet u de bewerking splitsen in twee afzonderlijke leesbewerkingen, één die eindigt op de tabelversie met de typewijziging en de andere vanaf de versie die de typewijziging bevat.

Limitations

Apache Iceberg-compatibiliteit

Apache Iceberg biedt geen ondersteuning voor alle typewijzigingen die worden gedekt door typebreiding. Zie Iceberg Schema Evolution.

Niet-ondersteunde typewijzigingen zijn onder andere:

  • byte, short, int, long naar decimal of double
  • Toename van de decimale schaal
  • date tot en met timestampNTZ

Wanneer u UniForm met Iceberg-compatibiliteit inschakelt voor een Delta Lake-tabel, resulteert het toepassen van een van de voorgaande typewijzigingen in een fout. Zie Delta Lake-tabellen lezen met Iceberg-clients met UniForm.

Als u een van deze niet-ondersteunde typewijzigingen toepast op een Delta Lake-tabel, hebt u twee opties:

  • Metagegevens van Iceberg opnieuw genereren: gebruik de volgende opdracht om icebergmetagegevens opnieuw te genereren zonder de functie voor het breder maken van tabellen:

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

    Hiermee kunt u uniformcompatibiliteit behouden nadat u incompatibele typewijzigingen hebt toegepast.

  • Schakel de functie voor type-uitbreidingstabel uit: Zie De functie voor type-uitbreidingstabel uitschakelen.

Typeafhankelijke functies

Sommige SQL-functies retourneren resultaten die afhankelijk zijn van het invoergegevenstype. De functie retourneert bijvoorbeeld hash verschillende hash-waarden voor dezelfde logische waarde als het argumenttype anders is: hash(1::INT) retourneert een ander resultaat dan hash(1::BIGINT).

Andere type afhankelijke functies zijn: xxhash64, bit_get, bit_reverse, typeof.

Voor stabiele resultaten in query's die gebruikmaken van deze functies, moet u expliciet waarden casten naar het gewenste type:

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

Niet-ondersteunde functies

  • U kunt een locatie voor het bijhouden van schema's niet instellen met behulp van SQL bij het streamen vanuit een Delta Lake-tabel met een typewijziging.
  • U kunt een tabel niet delen met type widening ingeschakeld voor niet-Databricks-gebruikers met behulp van OpenSharing.
  • Last updated on