Questa pagina fornisce informazioni sull'utilizzo della memoria nelle pipeline di Dataflow e sulla procedura per esaminare e risolvere i problemi relativi agli errori di esaurimento della memoria (OOM) di Dataflow.
Informazioni sull'utilizzo della memoria di Dataflow
Per risolvere i problemi di memoria insufficiente, è utile capire come le pipeline Dataflow utilizzano la memoria.
Quando Dataflow esegue una pipeline, l'elaborazione viene distribuita
su più macchine virtuali (VM) Compute Engine, spesso chiamate worker.
I worker elaborano gli elementi di lavoro dal servizio Dataflow
e delegare gli elementi di lavoro ai processi SDK Apache Beam. Un processo dell'SDK Apache Beam crea istanze di DoFn
. DoFn
è una classe SDK Apache Beam che definisce una funzione di elaborazione distribuita.
Dataflow avvia diversi thread su ogni worker e la memoria di ciascun worker è condivisa tra tutti i thread. Un thread è una singola attività eseguibile in esecuzione all'interno di un processo più grande. Il numero predefinito i thread dipendono da più fattori e variano tra job batch e flussi.
Se la pipeline ha bisogno di più memoria di quella predefinita disponibili sui worker, potresti riscontrare errori di memoria insufficiente.
Le pipeline Dataflow utilizzano principalmente la memoria dei worker in tre modi:
- Memoria operativa del worker
- Memoria di processo SDK
- Utilizzo della memoria di
DoFn
Memoria operativa del worker
I worker Dataflow hanno bisogno di memoria per i sistemi operativi e il sistema i processi di machine learning. La memoria utilizzata dai worker in genere non è superiore a 1 GB. L'utilizzo solitamente è inferiore a 1 GB.
- Vari processi sul worker utilizzano la memoria per garantire il corretto funzionamento della pipeline. Ciascuno di questi processi potrebbe prenotare una piccola quantità di memoria per il proprio funzionamento.
- Quando la pipeline non utilizza Streaming Engine, i processi di worker aggiuntivi utilizzano la memoria.
Memoria del processo SDK
I processi dell'SDK Apache Beam potrebbero creare oggetti e dati condivisi tra i thread all'interno del processo, indicati in questa pagina come oggetti e dati condivisi dell'SDK. L'utilizzo della memoria da parte di questi oggetti e dati condivisi dall'SDK è denominato memoria di processo dell'SDK. Il seguente elenco include esempi di oggetti e dati condivisi dall'SDK:
- Input aggiuntivi
- Modelli di machine learning
- Oggetti singleton in memoria
- Oggetti Python creati con il modulo
apache_beam.utils.shared
- Dati caricati da origini esterne, come Cloud Storage o BigQuery
I job di flussi che non utilizzano Streaming Engine archiviano gli input laterali in memoria. Per le pipeline Java e Go, ogni worker ha una copia dell'input aggiuntivo. Per Python di pipeline di input, ciascun processo dell'SDK Apache Beam ha una copia dell'input aggiuntivo.
I job di streaming che utilizzano Streaming Engine hanno un limite di dimensioni per gli input secondari di 80 MB. Gli input aggiuntivi sono archiviati al di fuori della memoria dei worker.
L'utilizzo della memoria da parte degli oggetti e dei dati condivisi dell'SDK cresce in modo lineare con il numero di processi dell'SDK Apache Beam. Nelle pipeline Java e Go, viene avviato un processo SDK Apache Beam per ogni worker. Nelle pipeline Python, viene avviato un processo dell'SDK Apache Beam per vCPU. Gli oggetti e i dati condivisi dell'SDK vengono riutilizzati nei thread all'interno dello stesso processo dell'SDK Apache Beam.
Utilizzo memoria di DoFn
DoFn
è una classe SDK Apache Beam che definisce una funzione di elaborazione distribuita.
Ogni worker può eseguire istanze DoFn
simultanee. Ogni thread esegue un DoFn
in esecuzione in un'istanza Compute Engine. Durante la valutazione dell'utilizzo totale della memoria, il calcolo delle dimensioni del set di lavoro o
necessaria per far funzionare un'applicazione, potrebbe essere
utili. Ad esempio, se un singolo DoFn
utilizza un massimo di 5 MB di memoria e un worker ha 300 thread, l'utilizzo della memoria di DoFn
potrebbe raggiungere un picco di 1,5 GB, ovvero il numero di byte di memoria moltiplicato per il numero di thread. A seconda di come i worker utilizzano la memoria, un picco di utilizzo della memoria potrebbe causare il loro esaurimento.
È difficile stimare quante istanze
DoFn
creati da Dataflow. Il numero dipende da vari fattori, come l'SDK, il tipo di macchina e così via. Inoltre, il DoFn potrebbe essere utilizzato da più thread in successione.
Il servizio Dataflow non garantisce quante volte viene invocato un DoFn
, né il numero esatto di istanze DoFn
create nel corso di una pipeline.
Tuttavia, la tabella seguente fornisce alcune informazioni sul livello
di parallelismo ci si può aspettare e che stima un limite superiore
il numero di istanze DoFn
.
SDK Beam Python
Batch | Streaming senza Streaming Engine | Streaming Engine | |
---|---|---|---|
Parallelismo |
1 processo per vCPU 1 thread per processo 1 thread per vCPU
|
1 processo per vCPU 12 thread per processo 12 thread per vCPU |
1 processo per vCPU 12 thread per processo 12 thread per vCPU
|
Numero massimo di istanze DoFn simultanee (tutti questi numeri sono soggetti a modifica in qualsiasi momento) |
1 DoFn per thread
1
|
1 DoFn per thread
12
|
1 DoFn per thread
12
|
SDK Beam Java/Go
Batch | Streaming senza Streaming Engine | Streaming Engine | |
---|---|---|---|
Parallelismo |
1 processo per VM worker 1 thread per vCPU
|
1 processo per VM worker 300 thread per processo 300 thread per VM worker
|
1 processo per VM worker 500 thread per processo 500 thread per VM worker
|
Numero massimo di istanze DoFn simultanee (tutti questi numeri sono soggetti a modifica in qualsiasi momento) |
1 DoFn per thread
1
|
1 DoFn per thread
300
|
1 DoFn per thread
500
|
Quando hai una pipeline multilingue e più di un SDK Apache Beam è in esecuzione sul worker, il worker utilizza il grado più basso possibile di parallelismo thread per processo.
Differenze tra Java, Go e Python
Java, Go e Python gestiscono i processi e la memoria in modo diverso. Di conseguenza, l'approccio che dovresti eseguire quando vuoi risolvere gli errori relativi all'esaurimento della memoria varia in base al indipendentemente dal fatto che la pipeline utilizzi Java, Go o Python.
Pipeline Java e Go
Nelle pipeline Java e Go:
- Ogni worker avvia un processo dell'SDK Apache Beam.
- Gli oggetti e i dati condivisi dall'SDK, come input secondari e cache, vengono condivisi tra tutti i thread del worker.
- La memoria utilizzata dagli oggetti e dai dati condivisi dell'SDK in genere non viene scalata in base al numero di vCPU sul worker.
Pipeline Python
Nelle pipeline Python:
- Ogni worker avvia un processo dell'SDK Apache Beam per vCPU.
- Gli oggetti e i dati condivisi dall'SDK, come gli input secondari e le cache, vengono condivisi tra tutti i thread all'interno di ogni processo dell'SDK Apache Beam.
- Il numero totale di thread sul worker scala in modo lineare in base al numero di vCPU. Di conseguenza, la memoria utilizzata da oggetti e dati condivisi dell'SDK aumenta in modo lineare con il numero di vCPU.
- I thread che eseguono il lavoro sono distribuiti tra i processi. Le nuove unità di lavoro vengono assegnate a un processo senza elementi di lavoro o al processo con il minor numero di elementi di lavoro attualmente assegnati.
Individuare gli errori di memoria insufficiente
Per determinare se la pipeline sta per esaurire la memoria, utilizza una uno dei seguenti metodi.
- Nella pagina Dettagli job, nel riquadro Log, visualizza la scheda Diagnostica. Questa scheda mostra gli errori relativi a problemi di memoria e la frequenza con cui si verificano.
- Nell'interfaccia di monitoraggio di Dataflow, utilizza il grafico Utilizzo della memoria per monitorare la capacità e l'utilizzo della memoria dei worker.
Nella pagina Dettagli job, nel riquadro Log, seleziona Log worker. Trova gli errori di memoria.
Java
Java Memory Monitor, configurato dal
Interfaccia di MemoryMonitorOptions
,
segnala periodicamente le metriche di garbage collection. Se la frazione di tempo di CPU utilizzata
La garbage collection supera una soglia del 50% per
per un periodo di tempo prolungato, l'attuale cablaggio dell'SDK ha esito negativo.
Potresti visualizzare un errore simile al seguente esempio:
Shutting down JVM after 8 consecutive periods of measured GC thrashing. Memory is used/total/max = ...
Questo errore di memoria può verificarsi quando la memoria fisica è ancora disponibile. L'errore di solito indica che l'utilizzo della memoria della pipeline è inefficiente. Per risolvere il problema, ottimizza la pipeline.
Se il tuo job ha un utilizzo elevato della memoria o errori di esaurimento della memoria, segui i consigli riportati in questa pagina per ottimizzare l'utilizzo della memoria o aumentare la quantità di memoria disponibile.
Risolvere gli errori di memoria insufficiente
Le modifiche alla pipeline di Dataflow potrebbero risolvere gli errori di esaurimento della memoria o ridurre l'utilizzo della memoria. Le possibili modifiche includono le seguenti azioni:
Il seguente diagramma mostra il flusso di lavoro per la risoluzione dei problemi di Dataflow descritto in questa pagina.
Ottimizzare la pipeline
Diverse operazioni della pipeline possono causare errori di esaurimento della memoria. Questa sezione fornisce opzioni per ridurre l'utilizzo della memoria della pipeline. a identificare la pipeline. fasi che consumano più memoria, utilizzare Cloud Profiler per monitorare le prestazioni della pipeline.
Per ottimizzare la pipeline, puoi utilizzare le seguenti best practice:
- Utilizzare i connettori I/O integrati di Apache Beam per leggere i file
- Riprogettare le operazioni quando si utilizzano
GroupByKey
PTransform - Riduci i dati in entrata da origini esterne
- Condividere oggetti tra thread
- Utilizzare rappresentazioni di elementi efficienti dalla memoria
- Riduci le dimensioni degli input laterali
Utilizza i connettori I/O integrati di Apache Beam per leggere i file
Non aprire file di grandi dimensioni all'interno di un file DoFn
. Per leggere i file, utilizza
Connettori I/O integrati Apache Beam.
I file aperti in un DoFn
devono rientrare nella memoria. Poiché vengono eseguite più istanze di DoFn
contemporaneamente, i file di grandi dimensioni aperti in DoFn
possono causare errori di esaurimento della memoria.
Riprogettare le operazioni quando si utilizzano le trasformazioni PTransform GroupByKey
Quando utilizzi un'istruzione PTransform GroupByKey
in Dataflow,
i valori per chiave e per finestra vengono elaborati su un singolo thread. Poiché questi dati
viene passato come flusso dal servizio di backend Dataflow
worker, non deve necessariamente
rientrare nella memoria dei worker. Tuttavia, se i valori sono
raccolti in memoria, la logica di elaborazione potrebbe causare errori di memoria insufficiente.
Ad esempio, se hai una chiave che contiene dati per una finestra e aggiungi i valori della chiave a un oggetto in memoria, ad esempio un elenco, potrebbero verificarsi errori di esaurimento della memoria. In questo caso, il worker potrebbe non avere una capacità di memoria sufficiente per contenere tutti gli oggetti.
Per ulteriori informazioni sulle PTransform GroupByKey
, consulta la documentazione di Apache Beam per GroupByKey
in Python e GroupByKey
in Java.
L'elenco seguente contiene suggerimenti per progettare la pipeline in modo da ridurre al minimo il consumo di memoria quando utilizzi le trasformazioni P GroupByKey
.
- Per ridurre la quantità di dati per chiave e finestra, evita le chiavi con molti valori, note anche come hot key.
- Per ridurre la quantità di dati raccolti per finestra, utilizza una finestra di dimensioni inferiori.
- Se utilizzi i valori di una chiave in una finestra per calcolare un numero, utilizza un
Trasformazione
Combine
: Non eseguire il calcolo in una singola istanza diDoFn
dopo aver raccolto i valori. - Filtra i valori o i duplicati prima dell'elaborazione. Per ulteriori informazioni, consulta la documentazione delle trasformazioni
Filter
per Python eFilter
per Java.
Riduci i dati in entrata da origini esterne
Se effettui chiamate a un'API esterna o a un database per l'arricchimento dei dati,
i dati restituiti devono rientrare
nella memoria del worker.
Se raggruppi le chiamate, ti consigliamo di utilizzare una trasformazione GroupIntoBatches
.
Se si verificano errori di memoria insufficiente, riduci la dimensione del batch. Per ulteriori informazioni sul raggruppamento in batch, consulta la documentazione delle trasformazioni GroupIntoBatches
per Python e GroupIntoBatches
per Java.
Condividere oggetti tra thread
La condivisione di un oggetto dati in memoria tra DoFn
istanze può migliorare lo spazio e
l'efficienza di accesso. Gli oggetti dati creati in qualsiasi metodo di DoFn
, inclusi
Setup
, StartBundle
, Process
, FinishBundle
e Teardown
, vengono invocati
per ogni DoFn
. In Dataflow, ogni worker potrebbe avere diversi DoFn
di Compute Engine. Per un utilizzo più efficiente della memoria, passa un oggetto dati come singleton per condividerlo tra più DoFn
. Per ulteriori informazioni, consulta il post del blog Riutilizzo della cache tra DoFn
.
Utilizza rappresentazioni degli elementi che consentono di risparmiare memoria
Valuta se puoi utilizzare rappresentazioni per gli elementi PCollection
che utilizzano meno memoria. Quando utilizzi programmatori nella tua pipeline, non considerare
solo rappresentazioni di elementi PCollection
codificate ma anche decodificate. Le matrici sparse possono spesso trarre vantaggio da questo tipo di ottimizzazione.
Riduci le dimensioni degli input aggiuntivi
Se i tuoi DoFn
utilizzano ingressi laterali, riduci le dimensioni dell'ingresso laterale. Da lato
ovvero raccolte di elementi, valuta l'utilizzo di viste iterabili, come
AsIterable
o AsMultimap
, invece di visualizzazioni che materializzano contemporaneamente l'intero input aggiuntivo, come
AsList
Rendi disponibile più memoria
Per aumentare la memoria disponibile, puoi aumentare la quantità totale di memoria disponibile sui worker senza modificare la quantità di memoria disponibile per thread. In alternativa, puoi aumentare la quantità di memoria disponibile per thread. Quando aumenti la memoria per thread, aumenti anche la memoria totale sul worker.
Puoi aumentare la quantità di memoria disponibile per thread in quattro modi:
- Utilizza un tipo di macchina con più memoria per vCPU.
- Utilizza un tipo di macchina con più vCPU (pipeline di flusso Java e Go).
- Riduci il numero di thread.
- Utilizza un solo processo SDK Apache Beam (flussi di dati Python e pipeline Python Runner v2).
Utilizza un tipo di macchina con più memoria per vCPU
Per selezionare un worker con più memoria per vCPU, utilizza uno dei seguenti metodi.
- Utilizza un tipo di macchina con memoria elevata nella famiglia di macchine per uso generico. I tipi di macchina con memoria elevata hanno una memoria per vCPU maggiore rispetto a dei tipi di macchina standard. L'utilizzo di un tipo di macchina con memoria elevata aumenta sia la memoria disponibile per ogni worker sia la memoria disponibile per thread, perché il numero di vCPU rimane invariato. Di conseguenza, l'utilizzo di un tipo di macchina con memoria elevata può essere un modo conveniente per selezionare un worker con più memoria per vCPU.
- Per una maggiore flessibilità nella specifica del numero di vCPU e della quantità memoria, puoi utilizzare un tipo di macchina personalizzata. Con i tipi di macchine personalizzate, puoi aumentare la memoria in incrementi di 256 MB. I prezzi di questi tipi di macchine sono diversi rispetto ai tipi di macchine standard.
- Alcune famiglie di macchine ti consentono di utilizzare tipi di macchine personalizzate con memoria estesa. La memoria estesa consente un rapporto memoria/vCPU più elevato. Il costo è più alto.
Per impostare i tipi di worker, utilizza la seguente opzione di pipeline. Per ulteriori informazioni, consulta Impostare le opzioni della pipeline e le Opzioni pipeline.
Java
Utilizza l'opzione pipeline --workerMachineType
.
Python
Utilizza l'opzione della pipeline --machine_type
.
Vai
Utilizza l'opzione della pipeline --worker_machine_type
.
Utilizza un tipo di macchina con più vCPU
Questa opzione è consigliata solo per le pipeline di flusso Java e Go. Tipi di macchina con più
Le vCPU hanno più memoria totale, perché la quantità di memoria scala in modo lineare
con il numero di vCPU. Ad esempio, un tipo di macchina n1-standard-4
con quattro
Le vCPU hanno 15 GB di memoria. Un tipo di macchina n1-standard-8
con otto vCPU ha 30 GB di memoria. Per ulteriori informazioni sui tipi di macchina predefinita, consulta
Famiglia di macchine per uso generico.
L'utilizzo di worker con un numero maggiore di vCPU potrebbe
aumentare notevolmente il costo della pipeline. Tuttavia, puoi utilizzare la scalabilità automatica orizzontale per ridurre il numero di worker totali in modo che il parallelismo rimanga invariato. Ad esempio, se hai 50 worker che utilizzano un tipo di macchina n1-standard-4
e passi a un tipo di macchina n1-standard-8
, puoi utilizzare la scalabilità automatica orizzontale e impostare il numero massimo di worker per ridurre il numero totale di worker nella pipeline a circa 25. Questa configurazione genera una pipeline con un costo simile.
Per impostare il numero massimo di worker, utilizza la seguente opzione della pipeline.
Java
Utilizza l'opzione della pipeline --maxNumWorkers
.
Per saperne di più, consulta Opzioni pipeline.
Vai
Utilizza l'opzione pipeline --max_num_workers
.
Per ulteriori informazioni, vedi Opzioni pipeline.
Questo metodo non è consigliato per le pipeline Python. Quando utilizzi l'SDK Python, se passi a un worker con un numero maggiore di vCPU, aumenti non solo la memoria, ma anche il numero di processi dell'SDK Apache Beam. Ad esempio, il tipo di macchina n1-standard-4
ha la stessa memoria per thread del tipo di macchina n1-standard-8
per le pipeline Python. Pertanto, con le pipeline Python,
si consiglia di utilizzare un tipo di macchina con memoria elevata, ridurre il numero di
thread o utilizzare un solo processo dell'SDK Apache Beam.
Riduci il numero di thread
Se l'utilizzo di un tipo di macchina con memoria elevata non risolve il problema, aumenta la quantità di memoria
disponibile per thread riducendo il numero massimo di thread che eseguono DoFn
istanze.
Questa modifica riduce il parallelismo. Per ridurre il numero di thread dell'SDK Apache Beam che eseguono istanze DoFn
, utilizza la seguente opzione di pipeline.
Java
Utilizza l'opzione della pipeline --numberOfWorkerHarnessThreads
.
Per saperne di più, consulta Opzioni pipeline.
Python
Utilizza l'opzione della pipeline --number_of_worker_harness_threads
.
Per saperne di più, consulta Opzioni pipeline.
Vai
Utilizza l'opzione pipeline --number_of_worker_harness_threads
.
Per saperne di più, consulta Opzioni pipeline.
Per ridurre il numero di thread per le pipeline batch Java e Go, imposta il valore
del flag a un numero inferiore al numero di vCPU sul worker. Per le pipeline in modalità flusso,
il valore del flag su un numero inferiore al numero di thread per
Processo SDK Apache Beam.
Per stimare i thread per processo, consulta la tabella nella sezione DoFn
Utilizzo della memoria di questa pagina.
Questa personalizzazione non è disponibile per le pipeline Python in esecuzione su Apache Beam SDK 2.20.0 o versioni precedenti o per le pipeline Python che non utilizzano Runner v2.
Utilizza un solo processo dell'SDK Apache Beam
Per le pipeline di flusso Python e le pipeline Python che utilizzano Runner v2, puoi obbliga Dataflow ad avviare un solo processo SDK Apache Beam per worker. Prima del giorno con questa opzione, prova innanzitutto a risolvere il problema utilizzando altri metodi. A configurare le VM worker Dataflow in modo da avviare una sola per il processo Python, utilizza il seguente opzione pipeline:
--experiments=no_use_multiple_sdk_containers
Con questa configurazione, le pipeline Python creano un solo processo SDK Apache Beam per worker. Questa configurazione impedisce che gli oggetti e i dati condivisi replicati più volte per ogni processo dell'SDK Apache Beam. Tuttavia, limita l'uso efficiente delle risorse di calcolo disponibili sul worker.
La riduzione del numero di processi dell'SDK Apache Beam a uno non necessariamente riduce il numero totale di thread avviati sul worker. Inoltre, avere tutti i thread in un unico processo dell'SDK Apache Beam potrebbe causare un'elaborazione lenta o bloccare la pipeline. Pertanto, potresti anche dover ridurre il numero di thread, come descritto nella sezione Ridurre il numero di thread di questa pagina.
Puoi anche forzare i worker a utilizzare un solo processo SDK Apache Beam utilizzando un tipo di macchina con una sola vCPU.