Inserire dati con il caricatore automatico

6 minuti

Quando i nuovi file di dati arrivano continuamente nell'archiviazione cloud, è necessario un modo efficiente per elaborarli senza tenere traccia manualmente dei file inseriti. Auto Loader risolve questa sfida rilevando e inserendo automaticamente nuovi file non appena compaiono, fornendo garanzie di esecuzione esatta una sola volta senza richiedere la gestione manuale dello stato o dei checkpoint.

Informazioni sul funzionamento del caricatore automatico

Il caricatore automatico è un'origine di Structured Streaming che monitora un percorso di archiviazione cloud e elabora in modo incrementale nuovi file. Invece di analizzare ripetutamente intere directory, il caricatore automatico mantiene lo stato sui file già elaborati, assicurandosi che ogni file venga inserito esattamente una volta.

Quando si avvia un flusso del caricatore automatico, può elaborare i file esistenti nella directory e quindi controllare continuamente i nuovi arrivi. Il flusso archivia le informazioni sullo stato di avanzamento in una posizione del checkpoint, che consente di riprendere esattamente da dove è stato arrestato se interrotto.

Il caricatore automatico rileva nuovi file usando una delle due modalità seguenti:

Modalità elenco directory: il caricatore automatico elenca periodicamente la directory di input per individuare nuovi file. Questo approccio non richiede alcuna configurazione aggiuntiva oltre l'accesso all'archiviazione.
Modalità di notifica file: il caricatore automatico usa servizi di notifica cloud per ricevere eventi all'arrivo di nuovi file. Questa modalità è più efficiente per i carichi di lavoro su larga scala perché evita elenchi di directory ripetuti.

Per la maggior parte dei carichi di lavoro di produzione, Databricks consiglia di abilitare la modalità di notifica file con eventi per file nella tua posizione esterna all'interno di Unity Catalog. Con gli eventi di file, il caricatore automatico riceve le notifiche direttamente all'arrivo dei file, riducendo così la latenza e i costi dell'API cloud.

Configurare il caricatore automatico per l'inserimento di file

Per inserire dati con il caricatore automatico, usate il formato cloudFiles con spark.readStream. L'esempio seguente legge i file JSON da Azure Data Lake Storage e li scrive in una tabella di Unity Catalog:

base_path = "abfss://container@storage.dfs.core.windows.net/autoloader/orders"
schema_path = f"{base_path}/schema"
checkpoint_path = f"{base_path}/checkpoint"

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", schema_path)
  .load("abfss://container@storage.dfs.core.windows.net/incoming/orders/")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("sales.bronze.orders"))

L'opzione cloudFiles.format specifica il formato del file di origine. Il caricatore automatico supporta file JSON, CSV, Parquet, Avro, ORC, XML, text e binari. Archivia cloudFiles.schemaLocation lo schema dedotto, che consente ad Auto Loader di tenere traccia delle modifiche dello schema nel corso del tempo. Anche se la posizione dello schema e il percorso del checkpoint possono essere uguali, mantenere separati consente di reimpostare un flusso senza perdere lo schema dedotto.

L'impostazione trigger(availableNow=True) elabora tutti i file attualmente disponibili e quindi si arresta. Questo modello funziona bene per i processi batch pianificati. Per l'elaborazione continua, è possibile omettere il trigger o usarlo trigger(processingTime="1 minute") per elaborare i file a intervalli.

Usare la sintassi SQL con "read_files"

Il caricatore automatico supporta anche la sintassi SQL tramite la read_files funzione con valori di tabella. Questo approccio è utile quando si preferisce SQL o quando si lavora nei notebook SQL:

SELECT * FROM read_files(
  'abfss://container@storage.dfs.core.windows.net/incoming/orders/',
  format => 'json',
  schemaHints => 'order_id INT, amount DECIMAL(10,2)'
)

Quando si usa read_files con le tabelle di streaming nelle pipeline di Lakeflow, le funzionalità di Auto Loader vengono abilitate automaticamente. Usare la keyword STREAM per abilitare l'elaborazione incrementale.

CREATE OR REFRESH STREAMING TABLE bronze_orders
AS SELECT * FROM STREAM read_files(
  'abfss://container@storage.dfs.core.windows.net/incoming/orders/',
  format => 'json'
)

In questo contesto, la pipeline gestisce automaticamente la memorizzazione dei punti di controllo e l'evoluzione dello schema.

Suggerimento

Quando si usa Unity Catalog, archivia i checkpoint e le posizioni dello schema in una posizione di archiviazione gestita. Ciò garantisce una governance coerente ed evita problemi di autorizzazione con i percorsi annidati.

Gestire l'inferenza e l'evoluzione dello schema

Il caricatore automatico può rilevare automaticamente lo schema dei file di origine, eliminando la necessità di definire manualmente gli schemi. Quando l'inferenza dello schema è abilitata, il caricatore automatico campiona i file per determinare i nomi di colonna e i tipi di dati.

Per i file JSON e CSV, il caricatore automatico deduce tutte le colonne come stringhe per impostazione predefinita. Questo approccio conservativo impedisce la mancata corrispondenza dei tipi che potrebbero causare la perdita di dati. Per abilitare il rilevamento automatico dei tipi, impostare l'opzione cloudFiles.inferColumnTypes :

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.inferColumnTypes", "true")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load(source_path))

Quando i dati di origine introduco nuove colonne, il caricatore automatico rileva la modifica e può evolvere automaticamente lo schema di destinazione. L'opzione cloudFiles.schemaEvolutionMode controlla questo comportamento:

Mode	Comportamento
`addNewColumns`	Il flusso si interrompe. Le nuove colonne vengono aggiunte allo schema. Al riavvio, il flusso continua con lo schema aggiornato. Si tratta dell'impostazione predefinita quando non viene fornito alcuno schema.
`addNewColumnsWithTypeWidening`	Il flusso si interrompe. Le nuove colonne vengono aggiunte allo schema e le modifiche al tipo di dati supportate , ad esempio `INT` a `LONG`, vengono ampliate automaticamente. Le modifiche del tipo non supportate vengono acquisite in `_rescued_data`. Richiede Databricks Runtime 16.4 o versione successiva.
`rescue`	Lo schema non è mai modificato e il flusso non fallisce. Tutte le nuove colonne vengono acquisite nella colonna `_rescued_data`.
`failOnNewColumns`	Il flusso si interrompe. Stream non viene riavviato a meno che lo schema specificato non venga aggiornato o che il file di dati in questione non venga rimosso.
`none`	Lo schema non è evoluto, le nuove colonne vengono ignorate e i dati non vengono salvati a meno che non sia impostata l'opzione `rescuedDataColumn` . Si tratta dell'impostazione predefinita quando viene fornito uno schema.

La colonna di dati salvata (_rescued_data) acquisisce tutti i dati che non corrispondono allo schema previsto, incluse le mancate corrispondenze di tipo e le colonne impreviste. Questa funzionalità impedisce la perdita di dati quando i dati di origine non sono conformi alle aspettative. Il caricatore automatico aggiunge automaticamente questa colonna durante l'inferenza dello schema.

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("cloudFiles.schemaEvolutionMode", "rescue")
  .schema(expected_schema)
  .load(source_path))

Per rinominare la colonna di dati salvata, usare l'opzione rescuedDataColumn :

.option("rescuedDataColumn", "my_rescued_data")

Se si sa che alcune colonne devono avere tipi specifici, usare hint dello schema per eseguire l'override dei tipi dedotti:

.option("cloudFiles.schemaHints", "order_date DATE, amount DECIMAL(10,2)")

Monitorare e ottimizzare i flussi del caricatore automatico

Per i carichi di lavoro di produzione, il monitoraggio consente di tenere traccia dello stato di avanzamento dell'inserimento e identificare i problemi. Il caricatore automatico fornisce una funzione SQL per eseguire query sullo stato dei file individuati:

SELECT * FROM cloud_files_state('/path/to/checkpoint');

Questa query restituisce i metadati relativi ai file individuati dal caricatore automatico, incluso il relativo stato di elaborazione. È possibile usare queste informazioni per verificare che i file vengano inseriti come previsto.

Il "Auto Loader" segnala le metriche tramite il dashboard di avanzamento di Structured Streaming, tra cui:

numFilesOutstanding: file individuati ma non ancora elaborati
numBytesOutstanding: Byte totali in attesa di elaborazione

Per carichi di lavoro con volumi elevati, controllare la frequenza di elaborazione usando cloudFiles.maxFilesPerTrigger o cloudFiles.maxBytesPerTrigger. Queste impostazioni impediscono che un singolo micro batch diventi troppo grande:

.option("cloudFiles.maxFilesPerTrigger", "1000")
.option("cloudFiles.maxBytesPerTrigger", "1g")

Importante

L'ubicazione del checkpoint archivia informazioni critiche sullo stato. Evitare di applicare i criteri del ciclo di vita dell'archiviazione cloud che potrebbero eliminare i file di checkpoint, in quanto questo danneggia lo stato del flusso e richiede il riavvio da zero.

Per accedere ai metadati a livello di file nei dati inseriti, selezionare la _metadata colonna . Questa colonna nascosta contiene informazioni su ogni file di origine, tra cui file_path, file_namefile_size, e file_modification_time:

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  .load(source_path)
  .select("*", "_metadata.file_path", "_metadata.file_modification_time"))

Auto Loader si integra perfettamente con le pipeline di Lakeflow, dove gestisce automaticamente il checkpoint e lo schema. Nell'unità successiva si apprenderà come usare il caricatore automatico nelle definizioni di pipeline dichiarative per i flussi di lavoro di inserimento dati end-to-end.

Commenti e suggerimenti

Questa pagina è stata utile?