Élargissement du type

Disponible sur les tables Delta Lake dans Databricks Runtime 15.4 LTS et versions ultérieures, l'élargissement de type vous permet de modifier les types de données de colonne vers un type plus large sans réécrire les fichiers de données.

Toutes les tables gérées par Unity Catalog utilisent Delta Lake par default. Consultez les tables gérées Unity Catalog pour Delta Lake et Apache Iceberg.

remarque

L'activation de l'élargissement de type met à jour les protocoles de lecture et d'écriture. Cela pourrait affecter la compatibilité avec les clients Delta Lake externes. Voir Compatibilité des fonctionnalités et protocoles de Delta Lake.

Les tables avec l'élargissement de type activé ne peuvent être lues que par Databricks Runtime 15.4 LTS et versions ultérieures.

Modifications de type prises en charge

Vous pouvez élargir les types selon les règles suivantes :

Type de source	Types plus larges pris en charge
`BYTE`	`SHORT`, `INT`, `BIGINT`, `DECIMAL`, `DOUBLE`
`SHORT`	`INT`, `BIGINT`, `DECIMAL`, `DOUBLE`
`INT`	`BIGINT`, `DECIMAL`, `DOUBLE`
`BIGINT`	`DECIMAL`
`FLOAT`	`DOUBLE`
`DECIMAL`	`DECIMAL` avec une plus grande précision et à une plus grande échelle
`DATE`	`TIMESTAMP_NTZ`
`VOID`	Tout type

Type de source	Types plus larges pris en charge
`BYTE`	`SHORT`, `INT`, `BIGINT`, `DECIMAL`, `DOUBLE`
`SHORT`	`INT`, `BIGINT`, `DECIMAL`, `DOUBLE`
`INT`	`BIGINT`, `DECIMAL`, `DOUBLE`
`BIGINT`	`DECIMAL`
`FLOAT`	`DOUBLE`
`DECIMAL`	`DECIMAL` avec une plus grande précision et à une plus grande échelle
`DATE`	`TIMESTAMP_NTZ`
`VOID`	Tout type

Les modifications de type sont prises en charge pour les colonnes de niveau supérieur et les champs imbriqués dans les structs, les maps et les tableaux.

remarque

VOID à tout type ne nécessite pas l'activation de l'élargissement du type sur la table. Toute opération qui met à jour le type d'une colonne VOID réussit sans configuration supplémentaire. L'élargissement de type VOID est disponible dans Databricks Runtime 18.2 et versions ultérieures.

Comportement décimal

Spark tronque la partie fractionnaire d'une valeur par default lorsqu'une opération promeut un type entier vers un decimal ou double et qu'une ingestion en aval réécrit la valeur dans une colonne de type entier. Pour plus de détails sur le comportement de la politique d'affectation, voir Affectation de magasin.

Lorsque vous modifiez un type numérique en decimal, la précision totale doit être égale ou supérieure à la précision de départ. Si vous augmentez également l'échelle, la précision totale doit augmenter d'un montant correspondant.

La cible minimale pour les types byte, short et int est decimal(10,0). La cible minimale pour long est decimal(20,0).

Si vous souhaitez ajouter deux décimales à un champ avec decimal(10,1), la cible minimale est decimal(12,3).

Activer l'élargissement de type

remarque

Vous pouvez activer l'élargissement des types sur une table existante en définissant la propriété de table delta.enableTypeWidening sur true:

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

Vous pouvez également activer l'élargissement de type lors de la création de table :

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

Appliquer manuellement un changement de type

Utilisez la commande ALTER COLUMN pour modifier les types manuellement :

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

Cette opération met à jour le schéma de la table sans réécrire les fichiers de données sous-jacents. Consultez ALTER TABLE pour plus de détails.

Élargir les types avec l'évolution automatique des schémas

Utilisez l'évolution des schémas avec l'élargissement de type pour mettre à jour les types de données dans les tables cibles afin qu'ils correspondent au type de données entrantes.

remarque

Sans l'élargissement de type activé, l'évolution des schémas tente toujours de réduire le type des données pour qu'elles correspondent aux types de colonnes de la table cible. Si vous ne souhaitez pas élargir automatiquement les types de données dans vos tables cibles, vous devez désactiver l’élargissement des types avant d’exécuter des charges de travail avec l’évolution des schémas activée.

Pour utiliser l'évolution des schémas afin d'élargir le type de données d'une colonne pendant l'ingestion, vous devez remplir les conditions suivantes :

La commande d'écriture s'exécute avec l'évolution automatique des schémas activée.
L'élargissement de type est activé sur la table cible.
Le type de colonne source est plus large que le type de colonne cible.
L'élargissement de type prend en charge la modification du type.

Les incompatibilités de type qui ne remplissent pas toutes ces conditions suivent les règles normales d'application des schémas. See application des schémas.

Exemple

Les exemples suivants démontrent comment l'élargissement de type fonctionne avec l'évolution des schémas.

Python
Scala
SQL

Créez une table cible avec une colonne INT et une table source avec une colonne BIGINT :

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

Utilisez saveAsTable() avec l'évolution des schémas pour élargir automatiquement la colonne INT à BIGINT lors d'une annexion :

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

Utilisez MERGE INTO avec l’évolution des schémas :

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

Créez une table cible avec une colonne INT et une table source avec une colonne BIGINT :

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

Utilisez saveAsTable() avec l'évolution des schémas pour élargir automatiquement la colonne INT à BIGINT lors d'une annexion :

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

Utilisez MERGE INTO avec l’évolution des schémas :

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

Créez une table cible avec une colonne INT et une table source avec une colonne BIGINT :

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

Utilisez INSERT INTO avec l'évolution des schémas pour élargir automatiquement la colonne INT à BIGINT lors d'une annexion :

SQL
INSERT WITH SCHEMA EVOLUTION INTO target_table SELECT * FROM source_table;

Utilisez MERGE INTO avec l’évolution des schémas :

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

Auto Loader

info

Aperçu

La prise en charge de l'élargissement de type dans Auto Loader est en aperçu public.

Auto Loader prend en charge l’élargissement de type avec l’évolution automatique des schémas. Lorsque vous utilisez Auto Loader pour ingérer des données dans une table Delta Lake avec l'élargissement des types et l'évolution des schémas activés, les types de colonnes sont automatiquement élargis pour correspondre aux données entrantes.

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

Voir élargissement automatique du type avec Auto Loader. De plus, l'élargissement de type doit être activé pour la table cible. Consultez Activer l'élargissement de type.

Désactiver la fonctionnalité d'élargissement de type de table

Vous pouvez empêcher l'élargissement de type accidentel sur les tables activées en définissant la propriété sur false:

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

Ce paramètre empêche les futurs changements de type de la table, mais ne supprime pas la fonctionnalité d'élargissement de type de la table et n'annule pas les changements de type précédents.

Si vous devez supprimer complètement les fonctionnalités d'élargissement de type de table, vous pouvez utiliser la commande DROP FEATURE comme indiqué dans l'exemple suivant :

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

remarque

Les tables qui ont permis l'élargissement de type à l'aide de Databricks Runtime 15.4 LTS vous obligent à abandonner la fonctionnalité typeWidening-preview à la place.

Lors de la suppression de l'élargissement de type, Databricks réécrit tous les fichiers de données qui ne sont pas conformes au schéma de table actuel. Voir supprimer une fonctionnalité de table Delta Lake et rétrograder le protocole de table.

Streaming à partir d'une table Delta Lake

La prise en charge de l'élargissement de type dans Structured Streaming est disponible dans Databricks Runtime 16.4 LTS et versions ultérieures.

Lorsque vous diffusez en continu à partir d'une table Delta Lake avec l'élargissement de type activé, vous pouvez configurer l'élargissement automatique de type pour les requêtes de streaming en activant l'évolution des schémas avec l'option mergeSchema sur la table cible. La table cible doit avoir l'élargissement de type activé. Voir Activer l'élargissement de type.

Python
Scala

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

Lorsque mergeSchema est activé et que la table cible a l'élargissement de type activé :

Les modifications de type sont appliquées automatiquement à la table en aval sans nécessiter d'intervention manuelle.
De nouvelles colonnes sont ajoutées automatiquement au schéma de table en aval.

Sans l'mergeSchema activé, les valeurs sont gérées selon la configuration spark.sql.storeAssignmentPolicy, qui par default rétrograde les valeurs pour correspondre au type de colonne cible. Pour plus d'informations sur le comportement de la politique d'affectation, consultez l'affectation de stockage.

Gérer les modifications de type dans un Stream

Lors du streaming à partir d'une table Delta Lake, vous pouvez fournir un emplacement de suivi du schéma pour suivre les modifications non additives du schéma, y compris les changements de type. La fourniture d'un emplacement de suivi de schéma est requise dans Databricks Runtime 18.0 et versions antérieures, et elle est facultative dans Databricks Runtime 18.1 et versions ultérieures.

Vous ne pouvez pas définir un schemaTrackingLocation à l'aide de SQL. Voir les fonctionnalités non prises en charge.

schemaTrackingLocation doit être défini sur un emplacement situé dans le même chemin que votre point de contrôle de streaming. Par exemple :

Python
Scala

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

Après avoir défini un emplacement de suivi de schéma, le Stream fait évoluer son schéma suivi lorsqu'il détecte un changement de type, puis s'arrête. À ce moment-là, vous devez gérer le changement de type, tel que l'activation de l'élargissement de type sur la table en aval ou la mise à jour de la query en streaming.

Pour reprendre le traitement, définissez la configuration Spark spark.databricks.delta.streaming.allowSourceColumnTypeChange ou l'option de lecteur DataFrame allowSourceColumnTypeChange, comme dans l'exemple suivant :

Python
Scala
SQL

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"

Lorsque le Stream s'arrête, un message d'erreur affiche l'ID de point de contrôle <checkpoint_id> et la version <delta_source_table_version> de la table source Delta Lake.

Pour une liste complète des options de streaming Delta Lake, consultez Delta Lake.

LakeFlow Pipelines

Vous pouvez activer l'élargissement de type pour les Lakeflow pipelines au niveau du pipeline ou pour des tables individuelles. L'élargissement des types permet d'élargir automatiquement les types de colonnes lors de l'exécution du pipeline sans nécessiter un refresh complet des tables de streaming. Les modifications de type dans les vues matérialisées trigger toujours un recalcul complet, et lorsqu'une modification de type est appliquée à une table source, les vues matérialisées qui dépendent de cette table nécessitent un recalcul complet pour refléter les nouveaux types.

Activer l’élargissement de type pour l’intégralité d’un pipeline

Pour activer l'élargissement de type pour toutes les tables d'un pipeline, définissez la configuration du pipeline pipelines.enableTypeWidening:

JSON
YAML

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

YAML
configuration:
  pipelines.enableTypeWidening: 'true'

Activer l'élargissement de type pour des tables spécifiques

Vous pouvez également activer l'élargissement de type pour les tables individuelles en définissant la propriété de table delta.enableTypeWidening:

Python
SQL

Python
import dlt

@dlt.table(
  table_properties={&quot;delta.enableTypeWidening&quot;: &quot;true&quot;}
)
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

Compatibilité avec les lecteurs en aval

Les tables avec élargissement de type activé peuvent être lues uniquement dans Databricks Runtime 15.4 LTS et versions ultérieures. Si vous souhaitez qu'une table avec élargissement de type activé dans votre pipeline soit lisible par les lecteurs sur Databricks Runtime 14.3 et versions antérieures, vous devez soit :

Désactivez l'élargissement de type en supprimant la propriété delta.enableTypeWidening/pipelines.enableTypeWidening ou en la définissant sur false, et Trigger un refresh complet de la table.
Activez le Mode de compatibilité sur votre table.

OpenSharing

remarque

La prise en charge de l'extension de type dans OpenSharing est disponible dans Databricks Runtime 16.1 et versions ultérieures.

Le partage d'une table Delta Lake avec l'élargissement de type activé est pris en charge dans OpenSharing de Databricks-to-Databricks. Le fournisseur et le destinataire doivent utiliser Databricks Runtime 16.1 ou version ultérieure.

Pour lire le flux de données de modification à partir d'une table Delta Lake avec l'élargissement de type activé en utilisant OpenSharing, vous devez définir le format de réponse sur delta:

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

La lecture du flux de données de modification en cas de changements de type n'est pas prise en charge. Vous devez plutôt diviser l'opération en deux lectures distinctes, l'une se terminant à la version de la table contenant la modification de type, et l'autre commençant à la version contenant la modification de type.

Limitations

Compatibilité Apache Iceberg

Apache Iceberg ne prend pas en charge toutes les modifications de type couvertes par l'élargissement de type. Voir l'évolution du schéma Iceberg.

Les modifications de type non prises en charge sont les suivantes :

byte, short, int, long à decimal ou double
augmentation de l'échelle décimale
date à la timestampNTZ

Lorsque vous activez UniForm avec la compatibilité Iceberg sur une table Delta Lake, l'application de l'une des modifications de type précédentes entraîne une erreur. Consultez lire les tables Delta Lake avec des clients Iceberg à l'aide de UniForm.

Si vous appliquez l'une de ces modifications de type non prises en charge à une table Delta Lake, vous avez deux options :

Régénérer les métadonnées Iceberg : utilisez la commande suivante pour régénérer les métadonnées Iceberg sans la fonctionnalité d'élargissement de type de table.
SQL
```
ALTER TABLE <table-name> SET TBLPROPERTIES ('delta.universalFormat.config.icebergCompatVersion' = '<version>')
```
Cela vous permet de maintenir une compatibilité uniforme après avoir appliqué des modifications de type incompatibles.
Supprimer la fonctionnalité d'élargissement de type de table : voir Désactiver la fonctionnalité d'élargissement de type de table.

Fonctions dépendantes du type

Certaines fonctions SQL renvoient des résultats qui dépendent du type de données d’entrée. Par exemple, la hash function renvoie des valeurs de hachage différentes pour une même valeur logique si le type d’argument est différent : hash(1::INT) renvoie un résultat différent de hash(1::BIGINT).

Les autres fonctions dépendantes du type sont : xxhash64, bit_get, bit_reverse, typeof.

Pour des résultats stables dans les requêtes qui utilisent ces fonctions, vous devez explicitement convertir les valeurs au type souhaité :

Python
Scala
SQL

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

Fonctionnalités non prises en charge

Vous ne pouvez pas définir un emplacement de suivi de schéma à l'aide de SQL lors du streaming depuis une table Delta Lake avec un changement de type.
Vous ne pouvez pas partager une table avec l'élargissement de type activé avec des consommateurs non-Databricks utilisant OpenSharing.

Modifications de type prises en charge​

Comportement décimal​

Activer l'élargissement de type​

Appliquer manuellement un changement de type​

Élargir les types avec l'évolution automatique des schémas​

Exemple​

Auto Loader​

Désactiver la fonctionnalité d'élargissement de type de table​

Streaming à partir d'une table Delta Lake​

Gérer les modifications de type dans un Stream​

LakeFlow Pipelines​

Activer l’élargissement de type pour l’intégralité d’un pipeline​

Activer l'élargissement de type pour des tables spécifiques​

Compatibilité avec les lecteurs en aval​

OpenSharing​

Limitations​

Compatibilité Apache Iceberg​

Fonctions dépendantes du type​

Fonctionnalités non prises en charge​

Modifications de type prises en charge

Comportement décimal

Activer l'élargissement de type

Appliquer manuellement un changement de type

Élargir les types avec l'évolution automatique des schémas

Exemple

Auto Loader

Désactiver la fonctionnalité d'élargissement de type de table

Streaming à partir d'une table Delta Lake

Gérer les modifications de type dans un Stream

LakeFlow Pipelines

Activer l’élargissement de type pour l’intégralité d’un pipeline

Activer l'élargissement de type pour des tables spécifiques

Compatibilité avec les lecteurs en aval

OpenSharing

Limitations

Compatibilité Apache Iceberg

Fonctions dépendantes du type

Fonctionnalités non prises en charge