Procedure consigliate per il caricatore automatico

Questa pagina descrive le best practice che puoi applicare per configurare Auto Loader affinché venga eseguito in modo affidabile, conveniente e scalabile per il tuo caso d'uso.

Queste best practice riducono il sovraccarico operativo e prevengono problemi comuni difficili da diagnosticare in produzione, come: costi API inutili LIST dovuti a scansioni complete delle directory, perdita silenziosa di dati dovuta alla deriva dello schema e riavvii della pipeline causati da una configurazione errata del checkpoint.

Per informazioni dettagliate sulla configurazione di produzione, vedere Configurare il caricatore automatico per i carichi di lavoro di produzione. Per il monitoraggio e l'osservabilità, vedere Monitorare e osservare il caricatore automatico.

Scegliere il framework di esecuzione corretto

Il framework di esecuzione migliore per il caso d'uso dipende dalla quantità di controllo necessaria per la pipeline e dal sovraccarico operativo che si vuole gestire. Per la maggior parte degli utenti e delle pipeline di produzione, il caricatore automatico con le pipeline Lakeflow è una soluzione ottimale. Tuttavia, se è necessario il controllo e la personalizzazione massimi, usare Il caricatore automatico con Structured Streaming. Per la configurazione più semplice con un'esperienza gestita, usare un connettore LakeFlow gestito, se disponibile.

Le pipeline di Lakeflow estendono Structured Streaming con scalabilità automatica, controlli di qualità dei dati, gestione dell'evoluzione dello schema e monitoraggio tramite il registro eventi. Databricks consiglia le pipeline Lakeflow per la maggior parte dei carichi di lavoro di acquisizione in produzione.

Scegli il tipo di pianificazione e di attivazione corretti

Il tipo di pianificazione e trigger migliore per il caso d'uso dipende dai requisiti di latenza e dai modelli di arrivo dei file. Per la maggior parte dei casi d'uso, Databricks consiglia un trigger di arrivo dei file con eventi di file abilitati. In questo modo si ottiene l'inserimento a bassa latenza a basso costo perché il calcolo viene eseguito solo quando arrivano nuovi file. I tre tipi di trigger differiscono in base alla frequenza e all'avvio della pipeline:

• Continuo: La pipeline viene eseguita senza interruzioni. Usare solo se una latenza inferiore al secondo è un requisito imprescindibile, poiché l’elaborazione continua costa di più. Associare gli eventi di file.
• Trigger all'arrivo di file: la pipeline inizia quando nuovi file arrivano nella posizione di origine. Ideale per la latenza da bassa a media o modelli di arrivo irregolari dei file. Richiede l'abilitazione degli eventi relativi ai file. Vedere Attivare i processi quando arrivano nuovi file.
• Pianificata: la pipeline viene eseguita secondo una pianificazione temporale (ad esempio, ogni ora). Usare quando i requisiti di latenza sono lenienti (da minuti a ore). Funziona con l'elenco di directory, ma gli eventi di file riducono i costi anche in modalità pianificata evitando analisi di directory complete.

Per informazioni dettagliate sull'uso Trigger.AvailableNow di per la pianificazione batch, vedere Uso di Trigger.AvailableNow e della limitazione della frequenza.

Scegliere la modalità di individuazione file corretta

Il caricatore automatico supporta tre modalità di individuazione file con compromessi diversi per la complessità, la scalabilità e i costi di configurazione.

Gli eventi di file consolidano le risorse di archiviazione cloud usando una sottoscrizione e una coda per ogni posizione esterna anziché una per flusso. La differenza di prestazioni è significativa su larga scala: l'elenco di directory deve analizzare l'intera directory di origine in ogni trigger, quindi il tempo di inserimento aumenta con le dimensioni della directory. Gli eventi di file recapitano direttamente nuove notifiche di file, quindi il tempo di inserimento rimane basso indipendentemente dal numero di oggetti presenti nella directory.

Abilitare gli eventi di file

Gli eventi sui file richiedono un'autorizzazione una tantum per il cloud e una posizione esterna configurata per utilizzare il servizio di eventi gestiti dei file. Dopo la configurazione, tutti i flussi del caricatore automatico che leggono da tale posizione esterna possono usare eventi di file senza alcuna configurazione aggiuntiva.

1. Concedere le autorizzazioni cloud necessarie sul lato del provider di servizi cloud. I requisiti variano in base al provider di servizi cloud. Consultare Configurare gli eventi di file in una posizione esterna.

2. Imposta cloudFiles.useManagedFileEvents su true nella query di Auto Loader.

   df = (spark.readStream
     .format("cloudFiles")
     .option("cloudFiles.format", "json")
     .option("cloudFiles.useManagedFileEvents", "true")
     .load("/path/to/data/dir"))
   

   Per la procedura di configurazione completa, consulta Eseguire la migrazione ad Auto Loader con eventi dei file.

Quando non è possibile usare gli eventi di file

Potrebbe non essere possibile usare gli eventi di file quando:

• La posizione esterna non è configurata con eventi relativi ai file.
• I criteri di sicurezza dell'organizzazione non consentono di abilitare gli eventi relativi ai file in una posizione esterna condivisa.

In questi casi, usare la modalità di notifica file classica o la modalità elenco directory. Per un confronto completo delle modalità di rilevamento dei file, vedere Confrontare le modalità di rilevamento file del caricatore automatico.

Gestire l'evoluzione dello schema

Auto Loader deduce automaticamente lo schema, ma il modo in cui si configura l'evoluzione dello schema influisce sulla completezza dei dati e sulla stabilità della pipeline. Usare la tabella seguente per scegliere una strategia.

Dopo aver scelto una strategia, applicare le procedure seguenti per ottimizzare il comportamento dell'evoluzione dello schema.

Usare hint dello schema per i tipi di campo noti

Usare l'opzione cloudFiles.schemaHints per applicare i tipi per i campi noti in anticipo, consentendo comunque l'inferenza dello schema per altri campi.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaHints", "id long, amount double")
  .load("/path/to/data/dir"))

Usare l'estensione dei tipi per le modifiche del tipo compatibili

La addNewColumnsWithTypeWidening modalità di evoluzione dello schema estende automaticamente i tipi compatibili (ad esempio, int a long) invece di instradare i dati alla _rescued_data colonna. In questo modo si evita la necessità di processi di post-elaborazione per gestire promozioni di tipi semplici.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  .option("cloudFiles.schemaEvolutionMode", "addNewColumnsWithTypeWidening")
  .load("/path/to/data/dir"))

Inserimento come Variant tipo per schemi imprevedibili

Quando i dati non sono conformi a uno schema specifico o lo schema cambia continuamente, inserire i dati come Variant tipo.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("singleVariantColumn", "data")
  .load("/path/to/data/dir"))

Variant fornisce la lettura dello schema in fase di query, ma è meno efficiente rispetto all'esecuzione di query su colonne strutturate. Per i meccanismi completi di inferenza ed evoluzione dello schema, vedere Configurare l'inferenza e l'evoluzione dello schema in Auto Loader.

Gestire dati non valido e qualità dei dati

Le procedure seguenti consentono di rilevare, acquisire e isolare dati non valido prima che vengano propagati ai livelli downstream.

Abilitare _rescued_data e _corrupt_record

Auto Loader offre due colonne per acquisire i dati che non possono essere analizzati correttamente.

• _rescued_data acquisisce campi che non corrispondono allo schema corrente. Viene aggiunto automaticamente dal caricatore automatico.
• _corrupt_record acquisisce righe che non possono essere analizzate affatto. Abilitarlo usando columnNameOfCorruptRecord:

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaHints", "_corrupt_record string")
  .option("columnNameOfCorruptRecord", "_corrupt_record")
  .load("/path/to/data/dir"))

Databricks consiglia columnNameOfCorruptRecord anziché badRecordsPath per evitare potenziali race condition che possono non rilevare record danneggiati.

Usa le aspettative nelle pipeline Lakeflow per il monitoraggio

Impostare le aspettative delle pipeline di Lakeflow per verificare che _rescued_data e _corrupt_record siano NULL in condizioni normali. I valori non NULL segnalano la deriva dello schema o il danneggiamento dei dati.

import dlt

@dlt.table
@dlt.expect("no rescued data", "_rescued_data IS NULL")
@dlt.expect("no corrupt records", "_corrupt_record IS NULL")
def bronze_table():
    return (spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaHints", "_corrupt_record string")
        .option("columnNameOfCorruptRecord", "_corrupt_record")
        .load("/path/to/data/dir"))

Isolare i dati danneggiati

Isolare le righe contenenti dati non analizzabili in un sink dedicato per l'analisi. In questo modo si impedisce la propagazione dei dati danneggiati ai livelli downstream.

import dlt

@dlt.table
def corrupt_records_sink():
    return dlt.read_stream("bronze_table").where("_corrupt_record IS NOT NULL")

@dlt.view
def clean_table():
    return dlt.read_stream("bronze_table").where("_corrupt_record IS NULL")

Annotare i dati con i metadati del file di origine

Includi la colonna _metadata nelle query di acquisizione di Auto Loader. Acquisire almeno file_path e file_modification_time. Ciò consente di ricondurre i problemi relativi ai dati a specifici file di origine e di fare join con cloud_files_state() per l’intero ciclo di vita dei file.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .load("/path/to/data/dir")
  .select("*", "_metadata.file_path", "_metadata.file_modification_time"))

Per informazioni dettagliate, vedere Colonna metadati file.

Ottimizzare i costi e le prestazioni

Le seguenti procedure riducono i tre principali fattori di costo di Auto Loader: chiamate API cloud LIST, risorse di calcolo inattive e crescita dello spazio di archiviazione a lungo termine.

• Usare gli eventi di file per ridurre al minimo LIST Costi API: gli eventi di file forniscono l'individuazione incrementale dei file, eliminando la necessità di elenchi di directory completi in ogni esecuzione. Questa è l'ottimizzazione dei costi con il maggiore impatto per Auto Loader.

• Usa i trigger di arrivo dei file per l'elaborazione basata su eventi: i trigger di arrivo dei file avviano la pipeline solo all'arrivo di nuovi file, quindi non paghi per risorse di calcolo inattive. Vedere Attivare i processi quando arrivano nuovi file.

• Archiviare i file elaborati con cloudFiles.cleanSource: usare cloudFiles.cleanSource per eliminare o spostare automaticamente i file elaborati. In questo modo si riducono sia i costi di archiviazione che i costi di elenco delle directory per i flussi di lunga durata. Per informazioni dettagliate, vedere Archiviazione dei file nella directory di origine per ridurre i costi.

  • Utilizzare la modalità delete per rimuovere i file dopo l'acquisizione.
  • Usare la modalità move per archiviare i file in un percorso diverso ai fini della conformità o della verifica.

  df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.cleanSource", "delete")
    .load("/path/to/data/dir"))
  

  Avvertimento

  Non abilitare cloudFiles.cleanSource se più stream di Auto Loader o altri client leggono dalla stessa directory di origine.

• Sfruttare i vantaggi dei miglioramenti delle prestazioni: eseguire l'aggiornamento alla versione più recente di Databricks Runtime o usare il calcolo serverless per trarre vantaggio dai recenti miglioramenti delle prestazioni del caricatore automatico.

Gestione dei checkpoint

Il checkpoint memorizza l'avanzamento del flusso e lo stato del file. La configurazione errata o la perdita del checkpoint richiede un riavvio completo, quindi considerarla come infrastruttura critica.

• Non applicare mai i criteri del ciclo di vita degli oggetti cloud alle posizioni dei checkpoint. Se i file di checkpoint vengono eliminati, lo stato del flusso è danneggiato ed è necessario riavviare da zero.
• Utilizzare checkpoint separati per ogni flusso e directory di origine.
• Prendere in considerazione cloudFiles.maxFileAge per flussi di volumi elevati di lunga durata per limitare la crescita dello stato. Usare un'impostazione conservativa (90 giorni minimo consigliato). Se si imposta questo valore in modo troppo aggressivo, si rischia di rielaborare i file già inseriti dal caricatore automatico se rientrano all'esterno della finestra.

Per informazioni dettagliate, vedere Rilevamento eventi di file.

Usare volumi per l'individuazione ottimale dei file con eventi di file

Per migliorare le prestazioni con gli eventi relativi ai file, creare un volume esterno per ogni percorso o sottodirectory da cui Auto Loader carica i file. Fornire percorsi di volume (ad esempio, /Volumes/catalog/schema/volume) al caricatore automatico anziché ai percorsi cloud (ad esempio, s3://bucket/path). In questo modo, l'individuazione dei file viene ottimizzata tramite un modello di accesso ai dati ottimizzato.

Per altre procedure consigliate per gli eventi di file, vedere Procedure consigliate per il caricamento automatico con eventi di file.