Configurez l'inférence et l'évolution du 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 les tables sans déclarer explicitement le schéma de données et de faire évoluer le schéma de la 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.
Auto Loader peut également "récupérer" des données inattendues (par exemple, de types de données différents) dans une colonne BLOB JSON, que vous pouvez choisir d'afficher ultérieurement à l'aide des APIs d'accès aux données semi-structurées.
Auto Loader prend en charge les formats suivants pour l'inférence et l'évolution du schéma :
Format de fichier | Versions prises en charge |
|---|---|
| Toutes les versions |
| Toutes les versions |
| Databricks Runtime 14.3 LTS et versions ultérieures |
| Databricks Runtime 10.4 LTS et versions ultérieures |
| Databricks Runtime 11.3 LTS ou version ultérieure |
| Non pris en charge |
| Non applicable (fixed-schema) |
| Non applicable (fixed-schema) |
Syntaxe pour l'inférence et l'évolution du schéma
La spécification d'un répertoire cible pour l'option cloudFiles.schemaLocation permet 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 les Lakeflow pipelines, Databricks gère automatiquement l'emplacement du schéma et d'autres informations de point de contrôle.
Si vous avez plus d'un emplacement de données source chargé dans la table cible, chaque charge de travail d'ingestion Auto Loader nécessite un point de contrôle de streaming distinct.
L'exemple suivant utilise parquet pour le cloudFiles.format. Utilisez csv, avro ou json pour les autres sources de fichiers. Tous les autres paramètres de lecture et d'écriture restent les mêmes pour les comportements par default de chaque format.
- Python
- 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>")
)
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 d'Auto Loader ?
Pour inférer le schéma lors de la première lecture des données, Auto Loader échantillonne les 50 Go ou 1 000 fichiers qu'il découvre, selon la limite atteinte en premier. Auto Loader stocke les information de schéma dans un répertoire _schemas à l'cloudFiles.schemaLocation configuré pour suivre les modifications de schéma des données d'entrée au fil du temps.
Pour modifier la taille de l'échantillon utilisé, définissez les configurations SQL :
spark.databricks.cloudFiles.schemaInference.sampleSize.numBytes
(chaîne d'octets, par exemple 10gb)
et
spark.databricks.cloudFiles.schemaInference.sampleSize.numFiles
(entier)
default, l'inférence de schémas d'Auto Loader vise à éviter les problèmes d'évolution des schémas dus à des incompatibilités de types. Pour les formats qui n'encodent pas les types de données (JSON, CSV et XML), Auto Loader infère toutes les colonnes en tant que 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 Merge les schémas des fichiers individuels. Le tableau suivant résume ce comportement.
Format de fichier | Type de données inféré par default |
|---|---|
| Chaîne |
| Chaîne |
| Chaîne |
| Types encodés dans le schéma Avro |
| Types encodés dans le schéma Parquet |
L’ Apache Spark DataFrameReader utilise un comportement différent pour l’inférence des schémas, en sélectionnant les types de données des colonnes dans les sources JSON, CSV et XML en fonction des données d’échantillon. Pour activer ce comportement avec Auto Loader, définissez l'option cloudFiles.inferColumnTypes sur true.
Lors de l'inférence du schéma pour les données CSV, Auto Loader suppose que les fichiers contiennent des en-têtes. Si vos fichiers CSV ne contiennent pas d'en-têtes, indiquez l'option .option("header", "false"). De plus, Auto Loader Merge les schémas de tous les fichiers de l'échantillon pour aboutir à un schéma global. Auto Loader peut ensuite lire chaque fichier selon son en-tête et analyser correctement le CSV.
Lorsqu'une colonne a des types de données différents dans deux fichiers Parquet, Auto Loader choisit le type le plus large. Vous pouvez utiliser schemaHints pour outrepasser ce choix. Lorsque vous spécifiez des indications de schéma, Auto Loader ne convertit pas la colonne au type spécifié, mais indique plutôt au lecteur Parquet de lire la colonne comme le type spécifié. En cas de non-concordance, Auto Loader récupère la colonne en plaçant les données dans la colonne de données récupérées.
Comment fonctionne l'évolution des schémas d'Auto Loader ?
Auto Loader détecte l'ajout de nouvelles colonnes lorsqu'il traite vos données. Quand Auto Loader détecte une nouvelle colonne, le Stream 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 dernier schéma 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 les streams Auto Loader avec Lakeflow Jobs pour redémarrer automatiquement après de telles modifications de schéma.
Auto Loader prend en charge les modes suivants pour l'évolution des schémas, que vous définissez dans l'option cloudFiles.schemaEvolutionMode :
Mode | Comportement lors de la lecture d'une nouvelle colonne |
|---|---|
| Le Stream échoue avec un |
| Même comportement que |
| Auto Loader ne fait jamais évoluer le schéma et le Stream ne échoue pas en raison des modifications du schéma. Auto Loader enregistre toutes les nouvelles colonnes dans la colonne de données sauvée. |
| Le Stream échoue et ne redémarre pas à moins que vous ne mettiez à jour le schéma fourni ou ne supprimiez le fichier de données incriminé. Le schéma n'est pas mis à jour automatiquement. |
| 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 |
addNewColumns le mode est le default lorsqu'aucun schéma n'est fourni, mais none est le default lorsque vous fournissez un schéma. addNewColumns n'est pas autorisé lorsque le schéma du Stream est fourni, mais fonctionne si vous fournissez votre schéma en tant qu'indication de schéma.
Auto Loader prend également en charge l'élargissement automatique des types avec le mode d'évolution des schémas addNewColumnsWithTypeWidening. Ce mode élargit automatiquement les types de données (tels que int à long ou float à double) sans nécessiter de réécriture des données ou d'intervention de l'utilisateur. Cette fonctionnalité est en Public Preview dans Databricks Runtime 16.4 et versions ultérieures. Consultez Élargissement automatique des types avec Auto Loader.
Comment les partitions fonctionnent-elles avec Auto Loader ?
Auto Loader tente d'inférer les colonnes de partition à partir de la structure de répertoire sous-jacente des données si les données sont organisées en 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épertoire sous-jacente contient des partitions Hive conflictuelles ou ne contient pas de partitionnement de style Hive, Auto Loader ignore les colonnes de partition.
Les formats de fichier binaire (binaryFile) et text ont des schémas de données fixes, mais prennent en charge l'inférence des colonnes de partition. Databricks recommande de configurer cloudFiles.schemaLocation pour ces formats de fichiers. Cela évite toute erreur potentielle ou perte d'information et empêche l'inférence des colonnes de partition chaque fois qu'un Auto Loader démarre.
Auto Loader ne tient pas compte des colonnes de partition pour l'évolution des schémas. Si vous aviez une structure de répertoires initiale comme base_path/event=click/date=2021-04-01/f0.json, et que vous commencez ensuite à recevoir de nouveaux fichiers sous la forme base_path/event=click/date=2021-04-01/hour=01/f1.json, Auto Loader ignore la colonne des heures. Pour capturer des informations pour de nouvelles colonnes de partition, définissez cloudFiles.partitionColumns sur event,date,hour.
L'option cloudFiles.partitionColumns accepte une liste de noms de colonnes séparés par des virgules. Auto Loader analyse uniquement les colonnes qui existent sous forme de paires key=value dans votre structure de répertoires.
Qu'est-ce que la colonne de données sauvée ?
Lorsque l'Auto Loader déduit le schéma, l'Auto Loader ajoute automatiquement une colonne de données récupérées à 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ées garantit qu'Auto Loader récupère les colonnes qui ne correspondent pas au schéma au lieu de les supprimer. La colonne de données sauvées contient toutes les données qui ne sont pas analysées pour les raisons suivantes :
- La colonne n'existe pas dans le schéma.
- Incohérences de type.
- Incohérences de casse.
La colonne de données récupérées contient un blob JSON avec les colonnes récupérées et le chemin du fichier source de l'enregistrement.
Les analyseurs JSON et CSV prennent en charge trois modes lors de l'analyse des enregistrements : PERMISSIVE, DROPMALFORMED et FAILFAST. Lorsqu'il est utilisé conjointement avec rescuedDataColumn, les incompatibilités de type de données n'entraînent pas la suppression d'enregistrements par Auto Loader en mode DROPMALFORMED ni le déclenchement d'une erreur en mode FAILFAST. Seuls les enregistrements corrompus échouent ou génèrent des erreurs, comme les JSON ou CSV incomplets ou malformés. Si vous utilisez badRecordsPath lors de l'analyse de JSON ou CSV, Auto Loader ne traite pas les incompatibilités de type de données comme de mauvais enregistrements lors de l'utilisation du rescuedDataColumn. Auto Loader stocke uniquement les enregistrements JSON ou CSV incomplets et malformés dans badRecordsPath.
Modifier le comportement sensible à la casse
Sauf si la sensibilité à la casse est activée, Auto Loader considère les colonnes abc, Abc et ABC comme étant la même colonne aux fins de l'inférence de schéma. Auto Loader choisit arbitrairement la casse en fonction des données échantillonnées. Vous pouvez utiliser des indications de schéma pour imposer la casse à utiliser. Une fois qu'Auto Loader a fait une sélection et inféré le schéma, il ne tient pas compte des variantes de casse qui n'ont pas été sélectionnées de manière cohérente avec le schéma.
Lorsque la colonne de données sauvées est activée, Auto Loader charge les champs nommés dans un cas autre que celui du schéma dans la colonne _rescued_data. Modifiez ce comportement en définissant l'option readerCaseSensitive sur « false », auquel cas Auto Loader lit les données d'une manière non sensible à la casse.
Remplacer la déduction de schéma par des indications de schéma
Vous pouvez utiliser des indications de schéma pour appliquer les informations de schéma que vous connaissez et attendez sur un schéma inféré. Lorsque vous savez qu'une colonne est d'un type de données spécifique, ou si vous souhaitez 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 la liste des types de données pris en charge, consultez Mappages de langues.
Si une colonne n'est pas présente au start du stream, vous pouvez également utiliser des indications de schéma pour ajouter cette colonne au schéma inféré.
L'exemple suivant montre un schéma inféré et le résultat de l'application des indications 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 indications de schéma suivantes :
.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 indications de schéma de tableau et de carte est disponible dans Databricks Runtime 9,1 LTS et versions ultérieures.
L'exemple suivant montre un schéma inféré avec des types de données complexes et le résultat de l'application des indications 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 indications de schéma suivantes :
.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
Auto Loader utilise les indications de schéma uniquement si vous ne fournissez pas de schéma. Vous pouvez utiliser les indications de schéma que cloudFiles.inferColumnTypes soit activé ou désactivé.