Risolvere gli errori di esaurimento della memoria di Dataflow

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

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 DoFn per vCPU

1 DoFn per thread

12 DoFn per vCPU

1 DoFn per thread

12 DoFn per vCPU

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 DoFn per vCPU

1 DoFn per thread

300 DoFn per VM worker

1 DoFn per thread

500 DoFn per VM worker

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.

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.

Un diagramma che mostra il flusso di lavoro per la risoluzione dei problemi.

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:

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 di DoFn 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 e Filter 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

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.