Configurare il caricatore automatico per i carichi di lavoro di produzione

Databricks consiglia di seguire le procedure consigliate per lo streaming per l'esecuzione del caricatore automatico nell'ambiente di produzione.

Databricks consiglia di usare il caricatore automatico nelle tabelle live Delta per l'inserimento incrementale dei dati. Le tabelle live delta estendono la funzionalità in Apache Spark Structured Streaming e consentono di scrivere solo alcune righe di Python dichiarativo o SQL per distribuire una pipeline di dati di qualità di produzione con:

• Scalabilità automatica dell'infrastruttura di calcolo per risparmiare sui costi
• Controlli di qualità dei dati con aspettative
• Gestione automatica dell'evoluzione dello schema
• Monitoraggio tramite metriche nel registro eventi

Monitoraggio del caricatore automatico

Esecuzione di query sui file individuati dal caricatore automatico

Il caricatore automatico fornisce un'API SQL per esaminare lo stato di un flusso. Usando la cloud_files_state funzione è possibile trovare i metadati relativi ai file individuati da un flusso del caricatore automatico. È sufficiente eseguire una query da cloud_files_state, specificando il percorso del checkpoint associato a un flusso del caricatore automatico.

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

Ascoltare gli aggiornamenti del flusso

Per monitorare ulteriormente i flussi del caricatore automatico, Databricks consiglia di usare l'interfaccia listener di query di streaming di Apache Spark.

Il caricatore automatico segnala le metriche al listener di query di streaming in ogni batch. È possibile visualizzare il numero di file presenti nel backlog e le dimensioni del backlog nelle numFilesOutstanding metriche e numBytesOutstanding nella scheda Dati non elaborati nel dashboard dello stato delle query di streaming:

{
  "sources" : [
    {
      "description" : "CloudFilesSource[/path/to/source]",
      "metrics" : {
        "numFilesOutstanding" : "238",
        "numBytesOutstanding" : "163939124006"
      }
    }
  ]
}

In Databricks Runtime 10.4 LTS e versioni successive, quando si usa la modalità di notifica file, le metriche includeranno anche il numero approssimativo di eventi di file presenti nella coda cloud come approximateQueueSize per AWS e Azure.

Considerazioni sui costi

Quando si esegue il caricatore automatico, l'origine principale dei costi sarebbe il costo delle risorse di calcolo e dell'individuazione dei file.

Per ridurre i costi di calcolo, Databricks consiglia di usare Processi di Databricks per pianificare il caricamento automatico come processi batch usando Trigger.AvailableNow invece di eseguirlo continuamente, purché non si disponga di requisiti di bassa latenza. Vedere Configurare gli intervalli di trigger di Structured Streaming.

I costi di individuazione dei file possono venire sotto forma di operazioni LIST negli account di archiviazione in modalità elenco directory e richieste API nel servizio di sottoscrizione e servizio di accodamento in modalità di notifica file. Per ridurre i costi di individuazione dei file, Databricks consiglia di:

• Specifica di un ProcessingTime trigger durante l'esecuzione del caricatore automatico in modo continuo in modalità elenco directory
• Progettazione dei caricamenti di file nell'account di archiviazione nell'ordinamento lessicale per sfruttare l'elenco incrementale (deprecato) quando possibile
• Utilizzo delle notifiche sui file quando non è possibile elencare incrementalmente
• Uso dei tag di risorsa per contrassegnare le risorse create dal caricatore automatico per tenere traccia dei costi

Uso di Trigger.AvailableNow e limitazione della frequenza

Il caricatore automatico può essere pianificato per l'esecuzione nei processi di Databricks come processo batch usando Trigger.AvailableNow. Il AvailableNow trigger indicherà al caricatore automatico di elaborare tutti i file arrivati prima dell'ora di inizio della query. I nuovi file caricati dopo l'avvio del flusso vengono ignorati fino al trigger successivo.

Con Trigger.AvailableNow, l'individuazione file avviene in modo asincrono con l'elaborazione dei dati e i dati possono essere elaborati in più micro batch con una limitazione della velocità. Il caricatore automatico per impostazione predefinita elabora un massimo di 1000 file ogni micro batch. È possibile configurare e cloudFiles.maxBytesPerTrigger per configurare cloudFiles.maxFilesPerTrigger il numero di file o il numero di byte da elaborare in un micro batch. Il limite di file è un limite rigido, ma il limite di byte è un limite flessibile, vale a dire che è possibile elaborare più byte rispetto all'oggetto fornito maxBytesPerTrigger. Quando le opzioni vengono fornite insieme, il caricatore automatico elabora il numero di file necessari per raggiungere uno dei limiti.

Conservazione degli eventi

Il caricatore automatico tiene traccia dei file individuati nel percorso del checkpoint usando RocksDB per fornire garanzie di inserimento esattamente una volta. Databricks consiglia vivamente di usare l'opzione cloudFiles.maxFileAge per tutti i flussi di inserimento di grandi volumi o di lunga durata. Questa opzione scade gli eventi dal percorso del checkpoint, che accelera il tempo di avvio del caricatore automatico. Il tempo di avvio può aumentare nei minuti per ogni esecuzione del caricatore automatico, che aggiunge costi non necessari quando si ha un limite superiore per l'età massima dei file che verranno archiviati nella directory di origine. Il valore minimo che è possibile impostare per cloudFiles.maxFileAge è "14 days". Le eliminazioni in RocksDB vengono visualizzate come voci di rimozione definitiva, pertanto è consigliabile che l'utilizzo dell'archiviazione aumenti temporaneamente man mano che gli eventi scadono prima di iniziare a livellarsi.

Se si tenta di ottimizzare l'opzione cloudFiles.maxFileAge , è possibile che i file non elaborati vengano ignorati dal caricatore automatico o che i file già elaborati scadono e quindi vengono rielaborati causando dati duplicati. Ecco alcuni aspetti da considerare quando si sceglie un oggetto cloudFiles.maxFileAge:

• Se il flusso viene riavviato dopo molto tempo, gli eventi di notifica dei file estratti dalla coda precedenti a cloudFiles.maxFileAge quelli ignorati. Analogamente, se si usa l'elenco di directory, i file che potrebbero essere apparsi durante il periodo di inattività precedenti cloudFiles.maxFileAge a quelli ignorati.
• Se si usa la modalità elenco directory e si usa cloudFiles.maxFileAge, ad esempio impostato su "1 month", si arresta il flusso e si riavvia il flusso con cloudFiles.maxFileAge impostato su "2 months", i file precedenti a 1 mese, ma più recenti di 2 mesi vengono rielaborati.

Se si imposta questa opzione la prima volta che si avvia il flusso, non verranno inseriti dati precedenti cloudFiles.maxFileAgea , pertanto, se si desidera inserire dati obsoleti, non è consigliabile impostare questa opzione quando si avvia il flusso per la prima volta. Tuttavia, è consigliabile impostare questa opzione nelle esecuzioni successive.

Attivare i backfill regolari usando cloudFiles.backfillInterval

Il caricatore automatico può attivare i backfill asincroni a un determinato intervallo, ad esempio un giorno per eseguire il backfill una volta al giorno o una settimana per il riempimento una volta alla settimana. I sistemi di notifica degli eventi file non garantiscono il 100% di recapito di tutti i file caricati e non forniscono contratti di servizio rigorosi sulla latenza degli eventi di file. Databricks consiglia di attivare normali backfill con il caricatore automatico usando l'opzione cloudFiles.backfillInterval per garantire che tutti i file vengano individuati all'interno di un determinato contratto di servizio se il completamento dei dati è un requisito. L'attivazione di backfill regolari non causa duplicati.