Configurare l'inferenza e l'evoluzione dello schema nell'Autoloader

È possibile configurare il caricatore automatico per rilevare automaticamente lo schema dei dati caricati, consentendo di inizializzare le tabelle senza dichiarare in modo esplicito lo schema dei dati ed evolvere lo schema della tabella man mano che vengono introdotte nuove colonne. In questo modo si elimina la necessità di tenere traccia e applicare manualmente le modifiche dello schema nel tempo.

Il caricatore automatico può anche "salvare" i dati imprevisti (ad esempio, di tipi di dati diversi) in una colonna BLOB JSON, che è possibile scegliere di accedere in un secondo momento usando le API di accesso ai dati semistrutturate.

Per l'inferenza e l'evoluzione dello schema sono supportati i formati seguenti:

Sintassi per l'inferenza e l'evoluzione dello schema

La specifica di una directory di destinazione per l'opzione cloudFiles.schemaLocation abilita l'inferenza dello schema e l'evoluzione. È possibile scegliere di usare la stessa directory specificata per checkpointLocation. Se si usano tabelle live Delta, Azure Databricks gestisce automaticamente la posizione dello schema e altre informazioni sul checkpoint.

Nell'esempio seguente viene parquet usato per .cloudFiles.format Usare csv, avroo json per altre origini file. Tutte le altre impostazioni per la lettura e la scrittura rimangono invariate per i comportamenti predefiniti per ogni formato.

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-checkpoint>")
  .load("<path-to-source-data>")
  .writeStream
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path_to_target")
)

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-checkpoint>")
  .load("<path-to-source-data>")
  .writeStream
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path_to_target")

Come funziona l'inferenza dello schema del caricatore automatico?

Per dedurre lo schema durante la prima lettura dei dati, il caricatore automatico campiona i primi 50 GB o 1000 file individuati, a qualunque limite venga superato per primo. Il caricatore automatico archivia le informazioni sullo schema in una directory _schemas configurata cloudFiles.schemaLocation per tenere traccia delle modifiche dello schema ai dati di input nel tempo.

spark.databricks.cloudFiles.schemaInference.sampleSize.numBytes

spark.databricks.cloudFiles.schemaInference.sampleSize.numFiles

Per impostazione predefinita, l'inferenza dello schema del caricatore automatico cerca di evitare problemi di evoluzione dello schema a causa di mancate corrispondenze del tipo. Per i formati che non codificano i tipi di dati (JSON, CSV e XML), Il caricatore automatico deduce tutte le colonne come stringhe (inclusi i campi annidati nei file JSON). Per i formati con schema tipizzato (Parquet e Avro), Il caricatore automatico esegue l'esempio di un subset di file e unisce gli schemi dei singoli file. Questo comportamento è riepilogato nella tabella seguente:

Apache Spark DataFrameReader usa un comportamento diverso per l'inferenza dello schema, selezionando i tipi di dati per le colonne in origini JSON, CSV e XML in base ai dati di esempio. Per abilitare questo comportamento con il caricatore automatico, impostare l'opzione cloudFiles.inferColumnTypes su true.

Come funziona l'evoluzione dello schema del caricatore automatico?

Il caricatore automatico rileva l'aggiunta di nuove colonne durante l'elaborazione dei dati. Quando il caricatore automatico rileva una nuova colonna, il flusso si arresta con un oggetto UnknownFieldException. Prima che il flusso generi questo errore, il caricatore automatico esegue l'inferenza dello schema nel micro batch di dati più recente e aggiorna il percorso dello schema con lo schema più recente unendo nuove colonne alla fine dello schema. I tipi di dati delle colonne esistenti rimangono invariati.

Databricks consiglia di configurare i flussi del caricatore automatico con i processi di Databricks per il riavvio automatico dopo tali modifiche dello schema.

Il caricatore automatico supporta le modalità seguenti per l'evoluzione dello schema, impostata nell'opzione cloudFiles.schemaEvolutionMode:

Come funzionano le partizioni con l'Autoloader?

Il caricatore automatico tenta di dedurre colonne di partizione dalla struttura di directory sottostante dei dati se i dati sono disposti nel partizionamento in stile Hive. Ad esempio, il percorso base_path/event=click/date=2021-04-01/f0.json del file determina l'inferenza di e event come colonne di date partizione. Se la struttura di directory sottostante contiene partizioni Hive in conflitto o non contiene il partizionamento dello stile Hive, le colonne di partizione vengono ignorate.

I file binari (binaryFile) e text i formati di file hanno schemi di dati fissi, ma supportano l'inferenza delle colonne di partizione. Databricks consiglia di impostare cloudFiles.schemaLocation per questi formati di file. In questo modo si evitano potenziali errori o perdite di informazioni e si impedisce l'inferenza delle colonne delle partizioni ogni volta che inizia un caricatore automatico.

Le colonne di partizione non vengono considerate per l'evoluzione dello schema. Se si dispone di una struttura di directory iniziale come base_path/event=click/date=2021-04-01/f0.jsone quindi iniziare a ricevere nuovi file come base_path/event=click/date=2021-04-01/hour=01/f1.json, Il caricatore automatico ignora la colonna dell'ora. Per acquisire informazioni per le nuove colonne di partizione, impostare su cloudFiles.partitionColumns event,date,hour.

Qual è la colonna di dati salvata?

Quando il caricatore automatico deduce lo schema, una colonna di dati salvata viene aggiunta automaticamente allo schema come _rescued_data. È possibile rinominare la colonna o includerla nei casi in cui si specifica uno schema impostando l'opzione rescuedDataColumn.

La colonna di dati salvata garantisce che le colonne che non corrispondono allo schema vengano salvate anziché essere eliminate. La colonna di dati salvata contiene tutti i dati che non vengono analizzati per i motivi seguenti:

• La colonna non è presente nello schema.
• Mancata corrispondenza del tipo.
• Mancata corrispondenza tra maiuscole e minuscole.

La colonna di dati salvata contiene un codice JSON contenente le colonne salvate e il percorso del file di origine del record.

Modificare il comportamento con distinzione tra maiuscole e minuscole

A meno che la distinzione tra maiuscole e minuscole non sia abilitata, le colonne abc, Abce ABC vengono considerate la stessa colonna ai fini dell'inferenza dello schema. Il caso scelto è arbitrario e dipende dai dati campionati. È possibile usare gli hint dello schema per applicare il caso da usare. Dopo aver effettuato una selezione e aver dedotto lo schema, il caricatore automatico non considera le varianti di maiuscole e minuscole non selezionate in modo coerente con lo schema.

Quando la colonna di dati salvata viene abilitata, i campi denominati in un caso diverso da quello dello schema vengono caricati nella _rescued_data colonna. Modificare questo comportamento impostando l'opzione readerCaseSensitive su false, nel qual caso il caricatore automatico legge i dati in modo senza distinzione tra maiuscole e minuscole.

Eseguire l'override dell'inferenza dello schema con hint dello schema

È possibile usare gli hint dello schema per applicare le informazioni sullo schema che si conoscono e si prevedono in uno schema dedotto. Quando si è certi che una colonna è di un tipo di dati specifico o se si vuole scegliere un tipo di dati più generale, ad esempio , double integerè possibile fornire un numero arbitrario di hint per i tipi di dati di colonna come stringa usando la sintassi della specifica dello schema SQL, ad esempio:

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

Vedere la documentazione sui tipi di dati per l'elenco dei tipi di dati supportati.

Se una colonna non è presente all'inizio del flusso, è anche possibile usare gli hint dello schema per aggiungere tale colonna allo schema dedotto.

Di seguito è riportato un esempio di schema dedotto per visualizzare il comportamento con gli hint dello schema.

Schema dedotto:

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

Specificando gli hint dello schema seguenti:

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

si ottiene:

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

Di seguito è riportato un esempio di schema dedotto con tipi di dati complessi per visualizzare il comportamento con gli hint dello schema.

Schema dedotto:

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

Specificando gli hint dello schema seguenti:

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

si ottiene:

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