Configuration de l’inférence et de l’évolution de schéma dans Auto Loader

Vous pouvez configurer Auto Loader pour détecter automatiquement le schéma des données chargées, ce qui vous permet d’initialiser des tables sans déclarer explicitement le schéma de données et faire évoluer le schéma de table à mesure que de nouvelles colonnes sont introduites. Cela élimine la nécessité de suivre et d’appliquer manuellement les modifications de schéma au fil du temps.

Le chargeur automatique peut également « sauver » les données inattendues (par exemple, des types de données différents) dans une colonne d’objet blob JSON, que vous pouvez choisir d’afficher ultérieurement à l’aide des API d’accès aux données semi-structurées.

Le chargeur automatique prend en charge les formats suivants pour l’inférence de schéma et l’évolution :

Syntaxe pour l’inférence de schéma et l’évolution

La spécification d’un répertoire cible pour l’option cloudFiles.schemaLocation active l’inférence et l’évolution du schéma. Vous pouvez choisir d’utiliser le même répertoire que celui que vous spécifiez pour le checkpointLocation. Si vous utilisez des pipelines Lakeflow, Azure Databricks gère automatiquement l’emplacement du schéma et d’autres informations de point de contrôle.

L’exemple suivant utilise parquet pour le cloudFiles.format. Utilisez csv, avro ou json pour d’autres sources de fichiers. Tous les autres paramètres de lecture et d’écriture restent les mêmes pour les comportements par défaut pour chaque format.

Python

(spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  # The schema location directory keeps track of your data schema over time
  .option("cloudFiles.schemaLocation", "<path-to-schema>")
  .load("<path-to-source-data>")
  .writeStream
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path-to-target>")
)

Langage de programmation Scala

spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  // The schema location directory keeps track of your data schema over time
  .option("cloudFiles.schemaLocation", "<path-to-schema>")
  .load("<path-to-source-data>")
  .writeStream
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path-to-target>")

Comment fonctionne l’inférence de schéma Auto Loader ?

Pour déduire le schéma lors de la première lecture de données, Auto Loader échantillonne les premiers 50 Go ou 1 000 fichiers qu’il découvre, selon la limite atteinte en premier. Auto Loader stocke les informations de schéma dans un répertoire _schemas au cloudFiles.schemaLocation configuré pour suivre les modifications apportées aux données d’entrée au fil du temps.

Pour modifier la taille de l’échantillon utilisé, définissez les configurations SQL suivantes :

• spark.databricks.cloudFiles.schemaInference.sampleSize.numBytes, une chaîne d’octets telle que 10gb.
• spark.databricks.cloudFiles.schemaInference.sampleSize.numFiles, un entier.

Par défaut, l’inférence de schéma Auto Loader cherche à éviter les problèmes d’évolution du schéma en raison d’incompatibilités de type. Pour les formats qui n’encodent pas les types de données (JSON, CSV et XML), le chargeur automatique déduit toutes les colonnes sous forme de chaînes (y compris les champs imbriqués dans les fichiers JSON). Pour les formats avec schéma typé (Parquet et Avro), Auto Loader échantillonne un sous-ensemble de fichiers et fusionne les schémas de fichiers individuels. Le tableau suivant récapitule ce comportement.

Le DataFrameReader Apache Spark utilise un comportement différent pour l’inférence de schéma, en sélectionnant des types de données pour les colonnes dans des sources JSON, CSV et XML en fonction des exemples de données. Pour activer ce comportement avec Auto Loader, définissez l’option cloudFiles.inferColumnTypes sur true.

Lors de l’inférence du schéma des données CSV, Auto Loader part du principe que les fichiers contiennent des en-têtes. Si vos fichiers CSV ne contiennent pas d’en-têtes, fournissez l’option .option("header", "false"). Auto Loader fusionne les schémas de tous les fichiers de l’échantillon pour créer un schéma global, puis lit chaque fichier selon son en-tête et analyse correctement le CSV.

Lorsqu’une colonne a différents types de données dans deux fichiers Parquet, Auto Loader choisit le type le plus large. Pour contourner ce choix, utilisez des indices de schéma. Lorsque vous spécifiez des indicateurs de schéma, le chargeur automatique ne convertit pas la colonne vers le type spécifié, mais indique au lecteur Parquet de la lire comme le type spécifié. Dans le cas d’une incompatibilité, le chargeur automatique sauve la colonne en plaçant les données dans la colonne de données sauvée.

Comment fonctionne l’évolution du schéma Auto Loader ?

Auto Loader détecte l’ajout de nouvelles colonnes lors du traitement de vos données. Lorsque le chargeur automatique détecte une nouvelle colonne, le flux s’arrête avec un UnknownFieldException. Avant que votre stream ne génère cette erreur, Auto Loader effectue l’inférence de schéma sur le dernier micro-batch de données et met à jour l’emplacement du schéma avec le schéma le plus récent en fusionnant les nouvelles colonnes à la fin du schéma. Les types de données des colonnes existantes restent inchangés.

Databricks recommande de configurer des flux Auto Loader avec Lakeflow Jobs pour redémarrer automatiquement après ces modifications de schéma.

Le chargeur automatique prend en charge les modes suivants pour l’évolution du schéma, que vous définissez dans l’option cloudFiles.schemaEvolutionMode :

Le chargeur automatique prend également en charge l’élargissement automatique des types avec le mode de schéma d’évolution addNewColumnsWithTypeWidening. Ce mode élargit automatiquement les types de données (par exemple int , vers long ou float vers double) sans nécessiter de réécriture de données ou d’intervention de l’utilisateur. Cette fonctionnalité est disponible en préversion publique dans Databricks Runtime 16.4 et versions ultérieures. Consultez la section Élargissement automatique du type avec Auto Loader.

Comment fonctionnent les partitions avec le chargeur automatique ?

Auto Loader tente de déduire les colonnes de partition à partir de la structure de répertoire sous-jacente des données si les données sont présentées dans un partitionnement de style Hive. Par exemple, le chemin de fichier base_path/event=click/date=2021-04-01/f0.json entraîne l'inférence de date et event comme colonnes de partition. Si la structure de répertoires sous-jacente contient des partitions Hive en conflit ou ne contient pas de partitionnement de style Hive, le chargeur automatique ignore les colonnes de partition.

Les formats de fichiers binaires (binaryFile) et text ont des schémas de données fixes, mais prennent en charge l'inférence de colonnes de partition. Databricks recommande de définir cloudFiles.schemaLocation pour ces formats de fichier. Cela évite toute erreur potentielle ou perte d’informations et empêche l’inférence des colonnes de partitions chaque fois qu’Auto Loader commence.

Le chargeur automatique ne prend pas en compte les colonnes de partition pour l’évolution du schéma. Si vous aviez une structure de répertoire initiale comme base_path/event=click/date=2021-04-01/f0.json, et que vous commencez ensuite à recevoir de nouveaux fichiers comme base_path/event=click/date=2021-04-01/hour=01/f1.json, Auto Loader ignore la colonne de l’heure. Pour capturer les informations des nouvelles colonnes de partition, définissez cloudFiles.partitionColumns sur une liste séparée par virgules des noms de colonnes, comme event,date,hour. Le chargeur automatique analyse uniquement les colonnes qui existent sous forme key=value de paires dans votre structure de répertoires.

Qu’est-ce que la colonne de données récupérées ?

Lorsque le chargeur automatique déduit le schéma, le chargeur automatique ajoute automatiquement une colonne de données sauvée à votre schéma en tant que _rescued_data. Vous pouvez renommer la colonne ou l’inclure lorsque vous fournissez un schéma en définissant l’option rescuedDataColumn .

La colonne de données sauvée garantit que le chargeur automatique sauve les colonnes qui ne correspondent pas au schéma au lieu de les supprimer. La colonne de données sauvée contient toutes les données qui ne sont pas analysées pour les raisons suivantes :

• La colonne est manquante dans le schéma.
• Incompatibilités de type.
• Incompatibilités de casse.

La colonne de données sauvée contient un objet blob JSON avec les colonnes sauvées et le chemin du fichier source de l’enregistrement.

Auto Loader sauve également des enregistrements composés entièrement de types de struct vides (structs avec champs zéro) dans des fichiers sources Avro et JSON. Parce que le format Parquet interdit les structures vides, ces enregistrements ne peuvent pas être écrits dans une table Delta en aval et sont redirigés vers _rescued_data à la place. Cela s’applique aux enregistrements Avro avec champs zéro et aux objets JSON avec champs zéro ({} au niveau supérieur ou comme valeur imbriquée mappée à une colonne struct).

Les analyseurs JSON et CSV prennent en charge trois modes d'analyse des enregistrements : PERMISSIVE, DROPMALFORMED, et FAILFAST. En cas d’utilisation avec rescuedDataColumn, les discordances de type de données n’entraînent pas de suppression d’enregistrements en mode DROPMALFORMED ou de génération d’erreur en mode FAILFAST par le chargeur automatique. Seuls les enregistrements endommagés échouent ou lèvent des erreurs, telles que des JSON ou CSV incomplets ou mal formés. Si vous utilisez badRecordsPath lors de l'analyse JSON ou CSV, le chargeur automatique ne traite pas les incompatibilités de type de données comme des enregistrements incorrects lors de l'utilisation du dispositif rescuedDataColumn. Le chargeur automatique stocke uniquement les enregistrements JSON ou CSV incomplets et mal formés dans badRecordsPath.

Modifier le comportement sensible à la casse

À moins que la sensibilité à la casse ne soit activée, le chargeur automatique considère que les colonnes abc, Abc et ABC sont la même colonne à des fins d’inférence de schéma. Le chargeur automatique choisit arbitrairement le cas en fonction des données échantillonées. Vous pouvez utiliser des indicateurs de schéma pour imposer la casse à utiliser. Une fois que le chargeur automatique a effectué une sélection et déduit le schéma, il ne considère pas les variantes de casse qui n’ont pas été sélectionnées comme conformes au schéma.

Lorsque la colonne de données sauvée est activée , le chargeur automatique charge les champs nommés dans un cas autre que celui du schéma dans la _rescued_data colonne. Modifiez ce comportement en définissant l’option readerCaseSensitive sur false, auquel cas le chargeur automatique lit les données sans respecter la casse.

Remplacer l’inférence de schéma par les indices de schéma

Vous pouvez utiliser des indications de schéma pour appliquer les informations de schéma que vous connaissez et que vous attendez d'un schéma déduit. Lorsque vous savez qu'une colonne est d'un type de données spécifique, ou si vous voulez choisir un type de données plus général (par exemple, un double au lieu d’un integer), vous pouvez fournir un nombre arbitraire d'indices pour les types de données de colonne sous forme de chaîne en utilisant la syntaxe de spécification de schéma SQL, comme suit :

.option("cloudFiles.schemaHints", "tags map<string,string>, version int")

Pour obtenir la liste des types de données pris en charge, consultez Correspondances de langages.

Si une colonne n’est pas présente au début du stream, vous pouvez également utiliser des indicateurs de schéma pour ajouter cette colonne au schéma déduit.

L’exemple suivant montre un schéma déduit et le résultat de l’application d’indicateurs de schéma.

Schéma déduit:

|-- date: string
|-- quantity: int
|-- user_info: struct
|    |-- id: string
|    |-- name: string
|    |-- dob: string
|-- purchase_options: struct
|    |-- delivery_address: string

En spécifiant les indicateurs de schéma suivants :

.option("cloudFiles.schemaHints", "date DATE, user_info.dob DATE, purchase_options MAP<STRING,STRING>, time TIMESTAMP")

Vous obtenez :

|-- date: string -> date
|-- quantity: int
|-- user_info: struct
|    |-- id: string
|    |-- name: string
|    |-- dob: string -> date
|-- purchase_options: struct -> map<string,string>
|-- time: timestamp

La prise en charge des indicateurs de schémas Array et Map est disponible dans Databricks Runtime 9.1 LTS et versions ultérieures.

L’exemple suivant montre un schéma déduit avec des types de données complexes et le résultat de l’application d’indicateurs de schéma.

Schéma déduit:

|-- products: array<string>
|-- locations: array<string>
|-- users: array<struct>
|    |-- users.element: struct
|    |    |-- id: string
|    |    |-- name: string
|    |    |-- dob: string
|-- ids: map<string,string>
|-- names: map<string,string>
|-- prices: map<string,string>
|-- discounts: map<struct,string>
|    |-- discounts.key: struct
|    |    |-- id: string
|    |-- discounts.value: string
|-- descriptions: map<string,struct>
|    |-- descriptions.key: string
|    |-- descriptions.value: struct
|    |    |-- content: int

En spécifiant les indicateurs de schéma suivants :

.option("cloudFiles.schemaHints", "products ARRAY<INT>, locations.element STRING, users.element.id INT, ids MAP<STRING,INT>, names.key INT, prices.value INT, discounts.key.id INT, descriptions.value.content STRING")

Vous obtenez :

|-- products: array<string> -> array<int>
|-- locations: array<int> -> array<string>
|-- users: array<struct>
|    |-- users.element: struct
|    |    |-- id: string -> int
|    |    |-- name: string
|    |    |-- dob: string
|-- ids: map<string,string> -> map<string,int>
|-- names: map<string,string> -> map<int,string>
|-- prices: map<string,string> -> map<string,int>
|-- discounts: map<struct,string>
|    |-- discounts.key: struct
|    |    |-- id: string -> int
|    |-- discounts.value: string
|-- descriptions: map<string,struct>
|    |-- descriptions.key: string
|    |-- descriptions.value: struct
|    |    |-- content: int -> string

Étapes suivantes

• Vérifier Auto Loader
• Configurer le chargeur automatique pour les charges de travail de production
• Explorer les modèles de chargement de données courants