Élargissement automatique du type avec Auto Loader

info

Aperçu

Cette fonctionnalité est en Aperçu public dans Databricks Runtime 16.4 et versions ultérieures.

Auto Loader traite de manière incrémentielle et efficace les nouveaux fichiers de données à mesure qu'ils arrivent dans le stockage cloud. Il réduit également la maintenance des pipelines en gérant automatiquement les modifications de schéma complexes. Par exemple, vous pouvez configurer Auto Loader pour détecter automatiquement le schéma des données chargées, ce qui vous permet d'initialiser les tables sans déclarer explicitement le schéma de données. Vous pouvez également faire évoluer le schéma de table à mesure que de nouvelles colonnes sont introduites, éliminant ainsi la nécessité de suivre et d'appliquer manuellement les modifications de schéma au fil du temps. Auto Loader peut même récupérer des données inattendues (par exemple, en raison de types de données différents) dans une colonne de données récupérées, ce qui vous aide à éviter la perte de données.

Cependant, la colonne de données récupérées exige que vous traitiez manuellement toute modification de type de données.

Pour gérer automatiquement certains de ces changements de type de données, utilisez l'élargissement de type dans Auto Loader. Delta Lake prend désormais en charge différentes modifications d'élargissement de type de données sans nécessiter de réécriture de données ou d'intervention de l'utilisateur. Voir Delta Lake Élargissement de type. Le nouveau mode pour l'évolution des schémas, addNewColumnsWithTypeWidening, fait automatiquement évoluer le schéma en cas de modifications de type de données compatibles.

Vous pouvez élargir les types primitifs comme int à long, float à double, et plus encore. L'élargissement de type est disponible pour tous les formats de fichier avec prise en charge de l'évolution des schémas dans Auto Loader. Cela inclut les formats texte (comme JSON, CSV ou XML) et les formats binaires (comme Avro ou Parquet). Il n’y a aucun changement dans le comportement d’évolution des schémas pour les modes d’évolution des schémas existants (tels que addNewColumns, rescue, failOnNewColumns ou none).

Modifications de type prises en charge

Les changements de type suivants sont pris en charge :

Type de source	Types plus larges pris en charge
byte	short, int, long, decimal, double
short	int, long, decimal, double
int	long, decimal, double
long	decimal
float	double
decimal	decimal avec une plus grande précision et à une plus grande échelle
date	timestampNTZ (uniquement pour les fichiers Parquet)

Type de source	Types plus larges pris en charge
byte	short, int, long, decimal, double
short	int, long, decimal, double
int	long, decimal, double
long	decimal
float	double
decimal	decimal avec une plus grande précision et à une plus grande échelle
date	timestampNTZ (uniquement pour les fichiers Parquet)

Lors de l'élargissement de tout type numérique à decimal, Auto Loader s'élargit à decimal avec une précision égale ou supérieure à la précision de départ. Si vous augmentez l'échelle, la précision totale augmente d'une quantité correspondante.

La précision de départ des types d'entiers est la suivante :

Type	Précision de départ.
byte	10
short	10
int	10
long	20

Type	Précision de départ.
byte	10
short	10
int	10
long	20

Par exemple, si le type actuel d'une colonne est int et qu'un fichier dont le type de colonne est decimal(5, 2) est lu, Auto Loader étend le type de cette colonne à decimal(12, 2).

Prérequis

Pour utiliser l'élargissement de type avec Auto Loader, vous devez satisfaire aux exigences suivantes :

• Utilisez Databricks Runtime 16.4 ou une version ultérieure.
• Si le récepteur d'écriture est une table Delta Lake, activez l'élargissement de type pour la table Delta Lake en utilisant l'une des méthodes suivantes :

  • Si vous utilisez une table existante :

    SQL

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

  • Si vous créez une nouvelle table avec l'élargissement de type activé :

    SQL

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

Pour plus d'informations sur l'élargissement de type dans les tables Delta Lake, consultez Élargissement de type.

Activer l'élargissement de type avec l'évolution des schémas

Pour utiliser l’élargissement de type avec Auto Loader, spécifiez addNewColumnsWithTypeWidening lors de l’utilisation de l’évolution des schémas. Auto Loader détecte l’ajout de nouvelles colonnes et les modifications de type au fur et à mesure qu’il traite vos données.

Python

Python

query = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("cloudFiles.inferColumnTypes", True)
  .option("cloudFiles.schemaLocation", <schemaPath>)
  .option("cloudFiles.schemaEvolutionMode", "addNewColumnsWithTypeWidening")
  .load(<inputPath>)
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", <checkpointPath>)
  .trigger(availableNow=True)
  .toTable("table_name")
)

Scala

Scala

val query = spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("cloudFiles.inferColumnTypes", true)
  .option("cloudFiles.schemaLocation", <schemaPath>)
  .option("cloudFiles.schemaEvolutionMode", "addNewColumnsWithTypeWidening")
  .load(<inputPath>)
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", <checkpointPath>)
  .trigger(Trigger.AvailableNow())
  .toTable("table_name")

Lorsque Auto Loader détecte une nouvelle colonne ou un changement de type pris en charge par l'élargissement de type, le Stream s'arrête avec un UnknownFieldException. Avant que votre Stream ne génère cette erreur, Auto Loader effectue une inférence de schéma sur le dernier micro-lot de données et met à jour l'emplacement du schéma avec le dernier schéma en élargissant les colonnes existantes ou en fusionnant de nouvelles colonnes à la fin du schéma.

Comportement d'évolution des schémas sur les modifications de type de données

Si vous deviez ingérer un CSV avec le contenu suivant, Auto Loader inférerait le schéma comme STRUCT<id INT, name STRING, _rescued_data STRING>.

CSV

id, name
1, John
2, Mary

La table cible se présente comme suit :

| id | name | _rescued_data | | -- | ---- | --------------- | | 1 | John | NULL | | 2 | Mary | NULL |

Maintenant, ingérez un autre fichier CSV où les valeurs de la colonne id sont plus larges que le type INT :

CSV

id, name, age
2147483648, Bob, 25

Le tableau suivant explique le comportement et le résultat avec différents modes d'évolution des schémas dans Auto Loader :

Mode	Comportement avec le changement de type de données extensible pris en charge
addNewColumns (default)	Le type de données n'évolue pas, et le Stream n'échoue pas en raison d'un changement de type de données. Les colonnes avec des valeurs dont les types ne correspondent pas sont définies sur NULL, et les valeurs non concordantes sont ajoutées à la colonne de données récupérées. Le Stream échoue en cas de nouvelles colonnes.
rescue	Le schéma n’évolue pas et les streams n’échouent pas en raison de modifications du schéma. Les colonnes avec des valeurs dont le type ne correspond pas sont définies sur NULL, et les valeurs incompatibles sont ajoutées à la colonne de données récupérées.
failOnNewColumns	Le type de données n'évolue pas, et le Stream n'échoue pas en raison d'un changement de type de données. Les colonnes avec des valeurs dont les types ne correspondent pas sont définies sur NULL, et les valeurs non concordantes sont ajoutées à la colonne de données récupérées. Le Stream échoue sur les nouvelles colonnes sans faire évoluer le schéma.
none	Ne fait pas évoluer le schéma, les nouvelles colonnes sont ignorées et les données ne sont pas récupérées à moins que l'option rescuedDataColumn ne soit définie. Stream ne échoue pas en raison de changements de schéma.
addNewColumnsWithTypeWidening	Le Stream échoue. De nouvelles colonnes sont ajoutées au schéma, et les modifications de type de données prises en charge sont élargies. Les modifications de type de données non prises en charge (par exemple, de int à string) sont ajoutées à la colonne de données récupérées.

Mode	Comportement avec le changement de type de données extensible pris en charge
addNewColumns (default)	Le type de données n'évolue pas, et le Stream n'échoue pas en raison d'un changement de type de données. Les colonnes avec des valeurs dont les types ne correspondent pas sont définies sur NULL, et les valeurs non concordantes sont ajoutées à la colonne de données récupérées. Le Stream échoue en cas de nouvelles colonnes.
rescue	Le schéma n’évolue pas et les streams n’échouent pas en raison de modifications du schéma. Les colonnes avec des valeurs dont le type ne correspond pas sont définies sur NULL, et les valeurs incompatibles sont ajoutées à la colonne de données récupérées.
failOnNewColumns	Le type de données n'évolue pas, et le Stream n'échoue pas en raison d'un changement de type de données. Les colonnes avec des valeurs dont les types ne correspondent pas sont définies sur NULL, et les valeurs non concordantes sont ajoutées à la colonne de données récupérées. Le Stream échoue sur les nouvelles colonnes sans faire évoluer le schéma.
none	Ne fait pas évoluer le schéma, les nouvelles colonnes sont ignorées et les données ne sont pas récupérées à moins que l'option rescuedDataColumn ne soit définie. Stream ne échoue pas en raison de changements de schéma.
addNewColumnsWithTypeWidening	Le Stream échoue. De nouvelles colonnes sont ajoutées au schéma, et les modifications de type de données prises en charge sont élargies. Les modifications de type de données non prises en charge (par exemple, de int à string) sont ajoutées à la colonne de données récupérées.

Résultats d'exemple

Les sections suivantes présentent le schéma inféré et les valeurs pour chaque mode d'évolution des schémas après l'ingestion du second fichier CSV.

addNewColumns

Schéma : id: INT, name: STRING, age: INT, _rescued_data: STRING

ID	Nom	âge	_rescued_data
1	John	NULL	NULL
2	Mary	NULL	NULL
NULL	Bob	25	{"id": 2147483648}

ID	Nom	âge	_rescued_data
1	John	NULL	NULL
2	Mary	NULL	NULL
NULL	Bob	25	{"id": 2147483648}

rescue

Schéma : id: INT, name: STRING, _rescued_data: STRING

ID	Nom	_rescued_data
1	John	NUL
2	Mary	NULL
NULL	Bob	{"age": 25, "id": 2147483648}

ID	Nom	_rescued_data
1	John	NUL
2	Mary	NULL
NULL	Bob	{"age": 25, "id": 2147483648}

failOnNewColumns

Schéma : id: IN_data: STRING

ID	Nom	_rescued_data
1	John	NULL
2	Mary	NULL
NULL	Bob	{"id": 2147483648}

ID	Nom	_rescued_data
1	John	NULL
2	Mary	NULL
NULL	Bob	{"id": 2147483648}

none

Schéma : `id: IN`

ID	Nom
1	John
2	Mary
NULL	Bob

ID	Nom
1	John
2	Mary
NULL	Bob

addNewColumnsWithTypeWidening

Schéma : id: BIGINT, name: STRING, age: INT, _rescued_data: STRING

ID	Nom	âge	_rescued_data
1	John	NULL	NULL
2	Mary	NULL	NULL
2 147 483 648	Bob	25	NULL

ID	Nom	âge	_rescued_data
1	John	NULL	NULL
2	Mary	NULL	NULL
2 147 483 648	Bob	25	NULL

Limitations

• L’option prefersDecimal ne peut pas être définie sur false lors de l’utilisation de addNewColumnsWithTypeWidening. Lorsque addNewColumnsWithTypeWidening est spécifié, la valeur par default de prefersDecimal est true.
• date l’élargissement à timestampNTZ n’est pris en charge que pour les fichiers Parquet.