La modalità di flessibilità avanzata (EFM) di Managed Service for Apache Spark gestisce i dati di shuffling per ridurre al minimo i ritardi nell'avanzamento dei job causati dalla rimozione dei nodi da un cluster in esecuzione. La modalità EFM trasferisce i dati di shuffling scrivendoli nei worker primari. I worker estraggono i dati da questi nodi remoti durante la fase di riduzione. Questa modalità è disponibile solo per i job Spark.
Poiché EFM non archivia i dati di shuffle intermedi sui worker secondari, è adatto ai cluster che utilizzano VM preemptible o eseguono la scalabilità automatica solo del gruppo di worker secondari.
EFM è supportato su Managed Service for Apache Spark
2.0.31+, 2.1.6+, 2.2+ e successive
versioni immagine.
- I job Apache Hadoop YARN che non supportano il trasferimento di AppMaster possono non riuscire in modalità di flessibilità avanzata (consulta Quando attendere il completamento di AppMaster).
- La modalità di flessibilità avanzata non è consigliata:
- su un cluster con soli worker primari
- sui job di flussi di dati, poiché possono essere necessari fino a 30 minuti dopo il completamento del job per ripulire i dati di shuffle intermedi.
- su un cluster che esegue notebook, poiché i dati di shuffle potrebbero non essere eliminati durante la sessione.
- quando i job Spark vengono eseguiti su un cluster con rimozione controllata controllata abilitata. La dismissione controllata e l'EFM possono avere scopi contrastanti, poiché il meccanismo di dismissione controllata di YARN mantiene i nodi IN FASE DI DISMISSIONE fino al completamento di tutte le applicazioni coinvolte.
- su un cluster che esegue job Spark e non Spark.
- La modalità di flessibilità avanzata non è supportata:
- quando è abilitata la scalabilità automatica dei worker primari. Nella maggior parte dei casi, i worker primari continueranno ad archiviare i dati di shuffle che non vengono migrati automaticamente. Lo scale down del gruppo di worker primari annulla i vantaggi della modalità EFM.
Utilizza la modalità di flessibilità avanzata
La flessibilità avanzata viene attivata quando crei un cluster impostando la
proprietà del cluster
dataproc:efm.spark.shuffle su primary-worker.
Esempio:
gcloud dataproc clusters create cluster-name \ --region=region \ --properties=dataproc:efm.spark.shuffle=primary-worker \ other flags ...
Esempio di Apache Spark
- Esegui un job WordCount sul testo pubblico di Shakespeare utilizzando il file jar degli esempi di Spark sul cluster EFM.
gcloud dataproc jobs submit spark \ --cluster=cluster-name \ --region=region \ --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \ --class=org.apache.spark.examples.JavaWordCount \ -- gs://apache-beam-samples/shakespeare/macbeth.txt
Configura le unità SSD locali
Poiché EFM scrive i dati di shuffling intermedi sui dischi collegati alle VM, trae vantaggio dal throughput e dalle IOPS aggiuntivi forniti dagli SSD locali. Per facilitare l'allocazione delle risorse, punta a un obiettivo di circa 1 partizione SSD locale ogni 4 vCPU durante la configurazione delle macchine worker principali.
Per collegare SSD locali, passa il flag --num-worker-local-ssds al comando gcloud Managed Service for Apache Spark clusters create.
In genere, non avrai bisogno di SSD locali sui worker secondari.
L'aggiunta di SSD locali ai worker secondari di un cluster (utilizzando il flag --num-secondary-worker-local-ssds) è spesso meno importante perché i worker secondari non scrivono i dati di shuffling localmente.
Tuttavia, poiché le unità SSD locali migliorano le prestazioni del disco locale, potresti decidere di aggiungerle ai worker secondari se prevedi che i job siano vincolati all'I/O a causa dell'utilizzo del disco locale: il job utilizza una quantità significativa di disco locale per lo spazio temporaneo o le partizioni sono troppo grandi per essere contenute nella memoria e verranno trasferite sul disco.
Rapporto worker secondari
Poiché i worker secondari scrivono i dati di shuffling sui worker primari, il tuo cluster deve contenere un numero sufficiente di worker primari con risorse di CPU, memoria e disco sufficienti per supportare il carico di shuffling del job. Per i cluster con scalabilità automatica, per impedire la scalabilità del gruppo primario e causare un comportamento indesiderato, imposta minInstances sul valore maxInstances nella policy di scalabilità automatica per il gruppo di worker primario.
Se il rapporto tra worker secondari e primari è elevato (ad esempio 10:1), monitora l'utilizzo della CPU, della rete e del disco dei worker primari per determinare se sono sovraccarichi. A tal fine:
Nella consoleGoogle Cloud , vai alla pagina Istanze VM.
Fai clic sulla casella di controllo a sinistra di worker primario.
Fai clic sulla scheda MONITORAGGIO per visualizzare l'utilizzo della CPU, le IOPS del disco, i byte di rete e altre metriche del worker primario.
Se i worker primari sono sovraccarichi, valuta la possibilità di eseguire lo scale up manuale dei worker primari.
Ridimensiona il gruppo di worker primari
Il gruppo di worker primari può essere sottoposto a scale up in sicurezza, ma lo scale down del gruppo di worker primari può influire negativamente sull'avanzamento del job. Le operazioni che riducono il gruppo di worker primari devono utilizzare la dismissione controllata, che viene attivata impostando il flag --graceful-decommission-timeout.
Cluster con scalabilità automatica: la scalabilità del gruppo di worker primari è disabilitata sui cluster EFM con policy di scalabilità automatica. Per ridimensionare il gruppo di worker primari in un cluster con scalabilità automatica:
Disabilita la scalabilità automatica.
gcloud dataproc clusters update \ --cluster=cluster-name \ --region=region \ --disable-autoscaling
Scalare il gruppo principale.
gcloud dataproc clusters update \ --cluster=cluster-name \ --region=region \ --num-workers=num-primary-workers \ --graceful-decommission-timeout=graceful-decommission-timeout # (if downscaling)
Riattiva la scalabilità automatica:
gcloud dataproc clusters update \ --cluster=cluster-name \ --region=region \ --autoscaling-policy=autoscaling-policy
Monitora l'utilizzo del disco dei worker primari
I worker primari devono avere spazio su disco sufficiente per i dati di shuffling del cluster.
Puoi monitorare questo aspetto indirettamente tramite la metrica remaining HDFS capacity.
Man mano che il disco locale si riempie, lo spazio diventa non disponibile per HDFS e la capacità rimanente diminuisce.
Per impostazione predefinita, quando l'utilizzo del disco locale di un worker primario supera il 90% della capacità, il nodo viene contrassegnato come NON INTEGRALE nell'interfaccia utente del nodo YARN. Se riscontri problemi di capacità del disco, puoi eliminare i dati inutilizzati da HDFS o fare lo scale up del pool di worker primari.
Configurazione avanzata
Partizionamento e parallelismo
Quando invii un job Spark, configura un livello di partizionamento appropriato. La scelta del numero di partizioni di input e output per una fase di shuffle comporta un compromesso tra diverse caratteristiche di prestazioni. È meglio sperimentare con valori adatti alle forme dei job.
Partizioni di input
Il partizionamento degli input di Spark e MapReduce è determinato dal set di dati di input. Durante la lettura dei file da Cloud Storage, ogni attività elabora circa una "dimensione del blocco" di dati.
Per i job Spark SQL, la dimensione massima della partizione è controllata da
spark.sql.files.maxPartitionBytes. Valuta la possibilità di aumentarla a 1 GB:spark.sql.files.maxPartitionBytes=1073741824.Per gli RDD Spark, la dimensione della partizione è solitamente controllata con
fs.gs.block.size, che per impostazione predefinita è 128 MB. Valuta la possibilità di aumentarla a 1 GB. Esempio:--properties spark.hadoop.fs.gs.block.size=1073741824
Partizioni di output
Il numero di attività nelle fasi successive è controllato da diverse proprietà. Per i job più grandi che elaborano più di 1 TB, ti consigliamo di avere almeno 1 GB per partizione.
Per Spark SQL, il numero di partizioni di output è controllato da
spark.sql.shuffle.partitions.Per i job Spark che utilizzano l'API RDD, puoi specificare il numero di partizioni di output o impostare
spark.default.parallelism.
Ottimizzazione dello shuffle per lo shuffle dei worker primari
La proprietà più significativa è --properties yarn:spark.shuffle.io.serverThreads=<num-threads>.
Tieni presente che si tratta di una proprietà YARN a livello di cluster perché il server di shuffle di Spark viene eseguito come parte di Node Manager. Il valore predefinito è il doppio (2x) del numero di core sulla macchina (ad esempio, 16 thread su un n1-highmem-8). Se "Tempo di blocco della lettura di shuffle" è superiore a 1 secondo e i worker primari non hanno raggiunto i limiti di rete, CPU o disco, valuta la possibilità di aumentare il numero di thread del server di shuffle.
Sui tipi di macchine più grandi, valuta la possibilità di aumentare spark.shuffle.io.numConnectionsPerPeer,
che per impostazione predefinita è 1. Ad esempio, impostalo su 5 connessioni per coppia di host.
Aumenta i nuovi tentativi
Il numero massimo di tentativi consentiti per master, attività e fasi dell'app può essere configurato impostando le seguenti proprietà:
yarn:yarn.resourcemanager.am.max-attempts spark:spark.task.maxFailures spark:spark.stage.maxConsecutiveAttempts
Poiché i master e le attività delle app vengono terminati più di frequente nei cluster che utilizzano molte VM preemptible o la scalabilità automatica senza rimozione controllata, l'aumento dei valori delle proprietà precedenti in questi cluster può essere utile (tieni presente che l'utilizzo di EFM con Spark e la rimozione controllata non è supportato).
Rimozione controllata di YARN sui cluster EFM
La dismissione controllata di YARN può essere utilizzata per rimuovere rapidamente i nodi con un impatto minimo sulle applicazioni in esecuzione. Per i cluster con scalabilità automatica, il timeout di dismissione controllata può essere impostato in un AutoscalingPolicy collegato al cluster EFM.
Miglioramenti di EFM alla rimozione controllata
Poiché i dati intermedi sono archiviati in un file system distribuito, i nodi possono essere rimossi da un cluster EFM non appena tutti i container in esecuzione su questi nodi hanno terminato. In confronto, i nodi non vengono rimossi sui cluster Managed Service for Apache Spark standard finché l'applicazione non è terminata.
La rimozione dei nodi non attende il completamento dei master dell'app in esecuzione su un nodo. Quando il container master dell'app viene terminato, viene ripianificato su un altro nodo che non è in fase di dismissione. L'avanzamento del job non viene perso: il nuovo master dell'app recupera rapidamente lo stato dal master dell'app precedente leggendo la cronologia dei job.