Scalabilità automatica dei cluster Dataproc

Che cos'è la scalabilità automatica?

La stima del "diritto" di worker (nodi) del cluster per un carico di lavoro è e una singola dimensione di cluster per un'intera pipeline spesso non è l'ideale. Il scaling del cluster avviato dall'utente risolve parzialmente questo problema, ma richiede il monitoraggio dell'utilizzo del cluster e l'intervento manuale.

L'API AutoscalingPolicies di Dataproc offre un meccanismo per automatizzare la gestione delle risorse del cluster e consente la scalabilità automatica delle VM worker del cluster. Autoscaling Policy è una configurazione riutilizzabile che descrive come scalare i worker del cluster che utilizzano il criterio di scalabilità automatica. Definisce limiti di scalabilità, frequenza e aggressività per fornire dati delle risorse del cluster per tutta la sua durata.

Quando utilizzare la scalabilità automatica

Utilizza la scalabilità automatica:

su cluster che archiviano i dati in servizi esterni, come Cloud Storage o BigQuery

su cluster che elaborano molti job

per eseguire lo scaling dei cluster con un singolo job

con la modalità di flessibilità avanzata per i job batch Spark

La scalabilità automatica non è consigliata con/per:

• HDFS: la scalabilità automatica non è concepita per scalare HDFS su cluster perché:

  1. L'utilizzo di HDFS non è un segnale per la scalabilità automatica.
  2. I dati HDFS sono ospitati solo sui worker principali. Il numero di istanze i worker devono essere sufficienti per ospitare tutti i dati HDFS.
  3. Il ritiro di DataNodes di HDFS può ritardare la rimozione dei worker. I datanodes copiano i blocchi HDFS in altri DataNodes prima che un worker venga rimosso. A seconda delle dimensioni dei dati e del fattore di replica, questa procedura può richiedere diverse ore.

• Etichette dei nodi YARN: la scalabilità automatica non supporta le Etichette dei nodi YARN né la proprietà dataproc:am.primary_only a causa del problema YARN-9088. YARN registra in modo errato le metriche del cluster quando vengono utilizzate le etichette dei nodi.

• Streaming strutturato Spark: la scalabilità automatica non supporta Streaming strutturato di Spark (vedi Scalabilità automatica e streaming strutturato Spark).

• Cluster inattivi:la scalabilità automatica non è consigliata per lo scopo di fare lo scale down di un cluster alla dimensione minima quando il cluster è inattivo. Poiché la creazione di un nuovo cluster è veloce quanto il ridimensionamento di uno esistente, ti consigliamo di eliminare i cluster inattivi e di ricrearli. Sono supportati dai seguenti strumenti "temporaneo" modello:

  Utilizzo di Dataproc Flussi di lavoro per pianificare un insieme di job su un cluster dedicato ed eliminare il cluster al termine dei job. Per un'orchestrazione più avanzata, utilizza Cloud Composer, basato su Apache Airflow.

  Per i cluster che elaborano ad hoc per le query o carichi di lavoro pianificati esternamente, Eliminazione pianificata dei cluster eliminare il cluster dopo un determinato periodo di inattività o una determinata durata in un determinato momento.

• Carichi di lavoro di dimensioni diverse: quando vengono eseguiti job di piccole e grandi dimensioni su un cluster, il ridimensionamento con disattivazione graduale attenderà il completamento dei job di grandi dimensioni. Il risultato è che un job a lunga esecuzione ritarda la scalabilità automatica per i job più piccoli in esecuzione sul cluster fino al job a lunga esecuzione finiture in cui sono finite. Per evitare questo risultato, raggruppa i job più piccoli di dimensioni simili su isola ogni job di lunga durata su un cluster a parte.

Attivazione della scalabilità automatica

Per attivare la scalabilità automatica su un cluster:

1. Crea un criterio di scalabilità automatica.

2. Una di queste soglie:

   1. Crea un cluster con la scalabilità automatica oppure
   2. Attiva la scalabilità automatica su un cluster esistente.

Crea un criterio di scalabilità automatica

workerConfig:
  minInstances: 10
  maxInstances: 10
secondaryWorkerConfig:
  maxInstances: 50
basicAlgorithm:
  cooldownPeriod: 4m
  yarnConfig:
    scaleUpFactor: 0.05
    scaleDownFactor: 1.0
    gracefulDecommissionTimeout: 1h

workerConfig:
  minInstances: 10
  maxInstances: 10
  weight: 1
secondaryWorkerConfig:
  minInstances: 0
  maxInstances: 100
  weight: 1
basicAlgorithm:
  cooldownPeriod: 2m
  yarnConfig:
    scaleUpFactor: 0.05
    scaleDownFactor: 1.0
    scaleUpMinWorkerFraction: 0.0
    scaleDownMinWorkerFraction: 0.0
    gracefulDecommissionTimeout: 1h

gcloud dataproc autoscaling-policies import policy-name \
    --source=filepath/filename.yaml \
    --region=region

Crea un cluster con la scalabilità automatica

Dopo aver creato un criterio di scalabilità automatica, crea un un cluster che utilizzerà il criterio di scalabilità automatica. Il cluster deve trovarsi nello stesso region come criterio di scalabilità automatica.

gcloud dataproc clusters create cluster-name \
    --autoscaling-policy=policy id or resource URI \
    --region=region

Attivare la scalabilità automatica in un cluster esistente

Dopo aver creato un criterio di scalabilità automatica, puoi attivarlo su un cluster esistente nella stessa regione.

gcloud dataproc clusters update cluster-name \
    --autoscaling-policy=policy id or resource URI \
    --region=region

Utilizzo dei criteri multi-cluster

• Un criterio di scalabilità automatica definisce il comportamento di ridimensionamento che può essere applicato a più cluster. Un criterio di scalabilità automatica viene applicato al meglio a più cluster quando i cluster condivideranno carichi di lavoro simili o eseguiranno job con un utilizzo delle risorse simile pattern.

• Puoi aggiornare un criterio utilizzato da più cluster. Aggiornamenti influisce immediatamente sul comportamento della scalabilità automatica per tutti i cluster che utilizzano (vedi autoscalingPolicies.update). Se non vuoi che un aggiornamento dei criteri venga applicato a un cluster che utilizza disabilita la scalabilità automatica sul cluster prima di aggiornare il criterio.

gcloud dataproc clusters update cluster-name --disable-autoscaling \
    --region=region

• Un criterio utilizzato da uno o più cluster non può essere eliminato (vedi autoscalingPolicies.delete).

Come funziona la scalabilità automatica

La scalabilità automatica controlla le metriche YARN Hadoop del cluster man mano che ogni periodo di "tempo di attesa" termina per determinare se eseguire lo scaling del cluster e, in caso affermativo, la portata dell'aggiornamento.

1. Il valore della metrica delle risorse YARN in attesa (Memoria in attesa o Core in attesa) determina se eseguire l'aumento o la riduzione di scala. Un valore maggiore di 0 indica che i job YARN sono in attesa di risorse e che potrebbe essere necessario aumentare le risorse. Un valore 0 indica che YARN ha risorse sufficienti per cui potrebbe non essere necessario un ridimensionamento o altre modifiche.

   Se la risorsa in attesa è maggiore di 0:

   $estimated\_worker\_count =$

   \[ \Biggl \lceil AVERAGE\ during\ cooldown\ period\Big(\frac{Pending + Available + Allocated + Reserved}{Resource\ per\ worker}\Big)\Biggr \rceil \]

   Se la risorsa in attesa è 0:

   $estimated\_worker\_count =$

   \[ \Biggl \lceil AVERAGE\ during\ cooldown\ period\Big(\frac{Allocated + Reserved}{Resource\ per\ worker}\Big)\Biggr \rceil \]

   Per impostazione predefinita, il gestore della scalabilità automatica monitora la risorsa di memoria YARN. Se attivi la scalabilità automatica basata su core, vengono monitorati sia la memoria YARN sia i core YARN:estimated_worker_count viene valutato separatamente per memoria e core e viene selezionato il numero di worker maggiore risultante.

   $estimated\_worker\_count =$

   \[ max(estimated\_worker\_count\_by\_memory,\ estimated\_worker\_count\_by\_cores) \]

   \[ estimated\ \Delta worker = estimated\_worker\_count - current\_worker\_count \]

2. Data la modifica stimata necessaria al numero di worker, la scalabilità automatica utilizza scaleUpFactor o scaleDownFactor per calcolare la variazione effettiva della numero di worker:

   if estimated Δworkers > 0:
     actual Δworkers = ROUND_UP(estimated Δworkers * scaleUpFactor)
     # examples:
     # ROUND_UP(estimated Δworkers=5 * scaleUpFactor=0.5) = 3
     # ROUND_UP(estimated Δworkers=0.8 * scaleUpFactor=0.5) = 1
   else:
     actual Δworkers = ROUND_DOWN(estimated Δworkers * scaleDownFactor)
     # examples:
     # ROUND_DOWN(estimated Δworkers=-5 * scaleDownFactor=0.5) = -2
     # ROUND_DOWN(estimated Δworkers=-0.8 * scaleDownFactor=0.5) = 0
     # ROUND_DOWN(estimated Δworkers=-1.5 * scaleDownFactor=0.5) = 0
   

   Un valore scaleUpFactor o scaleDownFactor pari a 1,0 indica che la scalabilità automatica verrà applicata in modo che la risorsa in attesa/disponibile sia pari a 0 (utilizzo perfetto).

3. Una volta calcolata la modifica del numero di worker, scaleUpMinWorkerFraction e scaleDownMinWorkerFraction fungono da soglie per determinare se la scalabilità automatica eseguirà lo scale del cluster. Una piccola frazione indica che la scalabilità automatica deve essere eseguita anche se Δworkers è piccolo. Una frazione più grande indica che la scalatura deve avvenire solo quando Δworkers è grande.

   IF (Δworkers >  scaleUpMinWorkerFraction * current_worker_count) then scale up
   

   OPPURE

   IF (abs(Δworkers) >  scaleDownMinWorkerFraction * current_worker_count),
   THEN scale down.
   

4. Se il numero di worker da scalare è sufficientemente elevato da attivare la scalabilità, la scalabilità automatica utilizza i limiti minInstances maxInstances di workerConfig e secondaryWorkerConfig e weight (rapporto tra worker principali e secondari) per determinare come suddividere il numero di worker tra le istanze principali e secondarie gruppi di istanza worker. Il risultato di questi calcoli è la modifica finale dell'autoscaling del cluster per il periodo di scaling.

5. Le richieste di riduzione della scalabilità automatica verranno annullate nei cluster creati con versioni delle immagini successive a 2.0.57 e 2.1.5 se:

   1. è in corso uno scale down con un valore del timeout per il ritiro gestito automaticamente diverso da zero e

   2. il numero di worker YARN ATTIVI ("lavoratori attivi") più la modifica del numero totale di worker consigliati dal gestore della scalabilità automatica (Δworkers) è uguale o superiore a DECOMMISSIONING worker YARN ("dismissione dei worker"), come illustrato nella seguente formula:

      IF (active workers + Δworkers ≥ active workers + decommissioning workers)
      THEN cancel the scaledown operation
      

   Per un esempio di annullamento dello scaledown, consulta Quando la scalabilità automatica annulla un'operazione di scaledown?.

Suggerimenti per la configurazione della scalabilità automatica

Evita di scalare i worker principali

I worker principali eseguono i datanode HDFS, mentre i worker secondari sono solo di calcolo. L'uso di worker secondari consente di scalare in modo efficiente le risorse di calcolo senza dalla necessità di eseguire il provisioning dello spazio di archiviazione, con conseguente aumento delle capacità di scalabilità. HDFS Namenode può avere più race condizioni che causano si danneggiano e si bloccano in modo indennizzo il ritiro. Per evitare questo problema, evita di scalare i worker principali. Ad esempio: workerConfig: minInstances: 10 maxInstances: 10 secondaryWorkerConfig: minInstances: 0 maxInstances: 100

Sono necessarie alcune modifiche alla creazione del cluster :

1. Imposta --num-workers=10 in modo che corrisponda al gruppo di worker principale del criterio di scalabilità automatica dimensioni.
2. Imposta --secondary-worker-type=non-preemptible per configurare i worker secondari non prerilasciabili. (a meno che non siano richieste VM prerilasciabili).
3. Copia la configurazione hardware dai worker principali ai worker secondari. Ad esempio, imposta --secondary-worker-boot-disk-size=1000GB in modo che corrisponda a --worker-boot-disk-size=1000GB.

Utilizzare la modalità di flessibilità avanzata per i job batch Spark

Utilizza la modalità di flessibilità avanzata (EFM) con la scalabilità automatica per:

Consente di ridurre più rapidamente il cluster mentre i job sono in esecuzione

Evitare interruzioni dei job in esecuzione a causa del ridimensionamento del cluster

riduci al minimo le interruzioni dei job in esecuzione a causa del prerilascio dei lavoratori secondari prerilasciabili

Con EFM abilitato, il criterio di scalabilità automatica il timeout per il ritiro deve essere impostato su 0s. Il criterio di scalabilità automatica deve con la scalabilità automatica dei worker secondari.

Scegliere un timeout per il ritiro gestito automaticamente

La scalabilità automatica supporta Ritiro gestito YARN durante la rimozione dei nodi da un cluster. Il ritiro gestito consente di presentare le applicazioni per completare lo shuffling dei dati tra le fasi per evitare di reimpostare l'avanzamento del job. Il ⁠timeout di ritiro graduale fornito in un criterio di scalabilità automatica è il limite superiore della durata per cui YARN attenderà le applicazioni in esecuzione (applicazioni in esecuzione all'avvio del ritiro) prima di rimuovere i nodi.

Quando un processo non viene completato entro il periodo di timeout per rimozione controllata specificato, il il nodo worker è stato arrestato in modo forzato, causando potenzialmente la perdita di dati o e un'interruzione del servizio. Per evitare questa possibilità, imposta il timeout per il ritiro graduale su un valore più lungo del il job più lungo elaborato dal cluster. Ad esempio, se prevedi che il job più lungo duri un'ora, imposta il timeout su almeno un'ora (1h).

Valuta la possibilità di eseguire la migrazione dei job che richiedono più di un'ora ai propri cluster temporanei per evitare di bloccare il ritiro graduale.

Impostazione di scaleUpFactor

scaleUpFactor controlla l'aggressività con cui il gestore della scalabilità automatica esegue lo scale up di un cluster. Specifica un numero compreso tra 0.0 e 1.0 per impostare il valore frazionario della risorsa in attesa di YARN che causa l'aggiunta di nodi.

Ad esempio, se ci sono 100 contenitori in attesa che richiedono ciascuno 512 MB, sono disponibili 50 GB di memoria YARN in attesa. Se scaleUpFactor è 0.5, il parametro il gestore della scalabilità automatica aggiungerà un numero di nodi sufficiente ad aggiungere 25 GB di memoria YARN. Allo stesso modo, se è 0.1, il gestore della scalabilità aggiungerà nodi sufficienti per 5 GB. Tieni presente che questi valori corrispondono alla memoria YARN, non alla memoria totale fisicamente disponibile su una VM.

Un buon punto di partenza è 0.05 per i job MapReduce e Spark con l'allocazione dinamica abilitata. Per i job Spark con un numero di esecutori fisso e job Tez, usa 1.0. Un valore scaleUpFactor pari a 1.0 indica che la scalabilità automatica verrà regolata in modo che la risorsa in attesa/disponibile sia pari a 0 (utilizzo perfetto).

Impostazione di scaleDownFactor

scaleDownFactor controlla l'aggressività con cui il gestore della scalabilità automatica fa lo scale down in un cluster Kubernetes. Specifica un numero compreso tra 0.0 e 1.0 per impostare il valore frazionario della risorsa YARN disponibile che causa la rimozione del nodo.

Lascia questo valore su 1.0 per la maggior parte dei cluster con più job che devono essere sottoposti a scale up e down di frequente. Come risultato del rimozione controllata, le operazioni di scale down molto più lento rispetto alle operazioni di scale up. Impostazione di scaleDownFactor=1.0 set una percentuale di scale down aggressiva, che riduce al minimo il numero di operazioni di scale down necessarie per raggiungere le dimensioni appropriate del cluster.

Per i cluster che richiedono una maggiore stabilità, imposta un valore scaleDownFactor inferiore per una frequenza di riduzione più lenta.

Imposta questo valore su 0.0 per impedire il ridimensionamento del cluster, ad esempio quando utilizzi cluster temporanei o con un solo job.

Configurazione di scaleUpMinWorkerFraction e scaleDownMinWorkerFraction

scaleUpMinWorkerFraction e scaleDownMinWorkerFraction sono in uso con scaleUpFactor o scaleDownFactor e hanno valori predefiniti valori di 0.0. Rappresentano le soglie a cui il gestore della scalabilità automatica espande o riduce il cluster: l'aumento o la diminuzione del valore frazionario minimo delle dimensioni del cluster necessario per emettere richieste di scale up o scale down.

Esempi: il gestore della scalabilità automatica non emette una richiesta di aggiornamento per aggiungere 5 worker a un cluster di 100 nodi, a meno che scaleUpMinWorkerFraction non sia inferiore o uguale a 0.05 (5%). Se il valore è impostato su 0.1, il gestore della scalabilità automatica non emetterà la richiesta di fare lo scale up del cluster. Allo stesso modo, se scaleDownMinWorkerFraction è 0.05, il gestore della scalabilità automatica non non inviare una richiesta di aggiornamento per rimuovere i nodi da un cluster da 100 nodi a meno che devono essere rimossi almeno cinque nodi.

Il valore predefinito 0.0 indica che non è impostata alcuna soglia.

Impostando una quota superiore scaleDownMinWorkerFractionthresholds su cluster di grandi dimensioni (> 100 nodi) per evitare piccole e non necessarie operazioni di scalabilità fortemente consigliato.

Scegliere un periodo di attesa

cooldownPeriod imposta un periodo di tempo durante il quale il gestore della scalabilità automatica non emetterà richieste di modifica delle dimensioni del cluster. Puoi utilizzarlo per limitare la frequenza delle modifiche del gestore della scalabilità automatica alle dimensioni del cluster.

Il valore minimo e predefinito cooldownPeriod è di due minuti. Se in un criterio viene impostato un valore cooldownPeriod più breve, il carico di lavoro cambia influenzerà più rapidamente le dimensioni dei cluster, ma questi potrebbero fare lo scale up inutilmente e giù. La pratica consigliata è impostare scaleUpMinWorkerFraction e scaleDownMinWorkerFraction di un criterio su un valore diverso da zero quando si utilizza un cooldownPeriod più breve. Ciò garantisce che fa lo scale up o lo scale down del cluster solo quando la variazione dell'utilizzo delle risorse sufficienti per garantire un aggiornamento del cluster.

Se il carico di lavoro è sensibile alle modifiche delle dimensioni del cluster, puoi aumentare il periodo di attesa. Ad esempio, se stai eseguendo un job di elaborazione collettiva, puoi impostare il periodo di attesa su almeno 10 minuti. Prova periodi di attesa diversi per trovare il valore più adatto al tuo carico di lavoro.

Limiti del conteggio di worker e pesi di gruppo

Ogni gruppo di worker ha minInstances e maxInstances che configurano un limite massimo per le dimensioni di ciascun gruppo.

Ogni gruppo ha anche un parametro chiamato weight che configura il target di equilibrio tra i due gruppi. Tieni presente che questo parametro è solo un suggerimento e che se un gruppo raggiunge le dimensioni minime o massime, i nodi verranno aggiunti o rimossi solo dall'altro gruppo. Pertanto, weight può quasi sempre essere lasciato al valore predefinito 1.

Abilita scalabilità automatica basata su core

Per impostazione predefinita, YARN utilizza le metriche di memoria per l'allocazione delle risorse. Per le applicazioni che richiedono un'elevata intensità di risorse della CPU, una best practice consiste nel configurare YARN in modo da utilizzare il Calcolatore delle risorse dominanti. Per farlo, imposta la seguente proprietà quando crei un cluster:

capacity-scheduler:yarn.scheduler.capacity.resource-calculator=org.apache.hadoop.yarn.util.resource.DominantResourceCalculator

Metriche e log di scalabilità automatica

Le risorse e gli strumenti seguenti possono aiutarti a monitorare le operazioni di scalabilità automatica e il loro effetto sul cluster e sui relativi job.

Cloud Monitoring

Utilizza Cloud Monitoring per:

• Visualizza le metriche utilizzate dalla scalabilità automatica
• Visualizza il numero di gestori di nodi nel cluster
• Scopri perché la scalabilità automatica ha o meno scalato il tuo cluster

Cloud Logging

Utilizza Cloud Logging per visualizzare i log di Cloud Dataproc Autoscaler.

1) Trova i log per il tuo cluster.

2) Seleziona dataproc.googleapis.com/autoscaler.

3) Espandi i messaggi di log per visualizzare il campo status. I log sono in formato JSON, un formato leggibile dal computer.

4) Espandi il messaggio di log per visualizzare i suggerimenti sulla scalabilità e le metriche utilizzate per decisioni di scalabilità, la dimensione originale del cluster e il nuovo cluster di destinazione dimensioni.

Informazioni di base: scalabilità automatica con Apache Hadoop e Apache Spark

Le sezioni seguenti illustrano in che modo la scalabilità automatica è (o meno) interoperabile con Hadoop YARN e Hadoop MapReduce, nonché con Apache Spark, Spark Streaming e Spark Structured Streaming.

Metriche YARN Hadoop

La scalabilità automatica si basa sulle seguenti metriche YARN Hadoop:

1. Allocated resource si riferisce alla risorsa YARN totale utilizzata dall'esecuzione di container in tutto il cluster. Se ci sono 6 container in esecuzione possono utilizzare fino a 1 unità di risorsa, sono state assegnate 6 risorse.

2. Available resource è la risorsa YARN nel cluster non utilizzata dai contenitori allocati. Se in tutti i gestori dei nodi sono presenti 10 unità di risorse e 6 vengono allocate, ci sono 4 risorse disponibili. Se nel cluster sono disponibili risorse (non utilizzate), la scalabilità automatica potrebbe rimuovere i worker dal cluster.

3. Pending resource è la somma delle richieste di risorse YARN per i container in attesa. I container in attesa sono in attesa di uno spazio per l'esecuzione in YARN. La risorsa in attesa è diverso da zero solo se la risorsa disponibile è pari a zero o troppo piccola per essere allocata containerizzato. Se sono presenti container in attesa, la scalabilità automatica potrebbe aggiungere worker al in un cluster Kubernetes.

Puoi visualizzare queste metriche in Cloud Monitoring. Per impostazione predefinita, la memoria YARN sarà 0,8 * memoria totale sul cluster, con di memoria rimanente riservata ad altri daemon e all'uso del sistema operativo, ad esempio la cache della pagina. Puoi ignorare il valore predefinito con l'impostazione di configurazione YARN "yarn.nodemanager.resource.memory-mb" (vedi Apache Hadoop YARN, HDFS, Spark e proprietà correlate).

Scalabilità automatica e Hadoop MapReduce

MapReduce esegue ogni mappa e riduce l'attività come container YARN separato. Quando il job inizia, MapReduce invia richieste al container per ogni attività di mappa, ottenendo un picco di memoria YARN in attesa. Al termine delle attività della mappa, la memoria in sospeso diminuisce.

Quando mapreduce.job.reduce.slowstart.completedmaps sono stati completati (95% per impostazione predefinita su Dataproc), MapReduce mette in coda le richieste dei contenitori per tutti i riduttore, con un altro picco di memoria in attesa.

A meno che le attività di mappatura e riduzione non richiedano diversi minuti o più, non impostare un valore elevato per la scalabilità automatica scaleUpFactor. Aggiunta di worker a richiede almeno un minuto e mezzo, quindi assicurati che lavoro in sospeso sufficiente per utilizzare un nuovo worker per alcuni minuti. Un buon inizio punto è impostare scaleUpFactor su 0,05 (5%) o 0,1 (10%) di memoria in sospeso.

Scalabilità automatica e Spark

Spark aggiunge un ulteriore livello di pianificazione a YARN. Nello specifico, L'allocazione dinamica invia richieste a YARN affinché i container eseguano gli esecutori Spark, quindi pianifica le attività Spark sui thread di questi esecutori. Cluster Dataproc abilitare l'allocazione dinamica per impostazione predefinita, in modo che gli esecutori vengano aggiunti e rimossi in base alle esigenze.

Spark chiede sempre i container a YARN, ma senza l'allocazione dinamica, richiede i container all'inizio del job. Con l'allocazione dinamica,rimuoverà i container o ne richiederà di nuovi, se necessario.

Spark parte da un numero limitato di esecutori, 2 su cluster con scalabilità automatica, e continua a raddoppiare il numero di esecutori mentre ci sono attività in backlog. In questo modo, la memoria in attesa viene uniformata (meno picchi di memoria in attesa). Per i job Spark, ti consigliamo di impostare la scalabilità automatica scaleUpFactor su un numero elevato, ad esempio 1,0 (100%).

Disabilitazione dell'allocazione dinamica di Spark

Se esegui job Spark separati che non traggono vantaggio dalla creatività dinamica Spark puoi disabilitare l'allocazione dinamica di Spark impostando spark.dynamicAllocation.enabled=false e impostazione spark.executor.instances. Puoi comunque utilizzare la scalabilità automatica per aumentare e diminuire i cluster durante l'esecuzione dei job Spark separati.

Job Spark con dati memorizzati nella cache

Imposta spark.dynamicAllocation.cachedExecutorIdleTimeout o annulla la memorizzazione nella cache dei set di dati quando non sono più necessari. Per impostazione predefinita Spark non rimuove gli esecutori che hanno i dati memorizzati nella cache, il che impedirebbe lo scale down del cluster.

Scalabilità automatica e Spark Streaming

1. Poiché Spark Streaming ha una propria versione dell'allocazione dinamica che utilizza indicatori specifici per lo streaming per aggiungere e rimuovere gli esecutori, imposta spark.streaming.dynamicAllocation.enabled=true e disattiva l'allocazione dinamica di Spark Core impostando spark.dynamicAllocation.enabled=false.

2. Non utilizzare il ritiro gestito automaticamente (scalabilità automatica gracefulDecommissionTimeout) con i job Spark Streaming. Per rimuovere in sicurezza i worker con la scalabilità automatica, configura il checkpointing per la tolleranza ai guasti.

In alternativa, per utilizzare Spark Streaming senza la scalabilità automatica:

1. Disattivare l'allocazione dinamica di Spark Core (spark.dynamicAllocation.enabled=false) e
2. Imposta il numero di esecutori (spark.executor.instances) per un lavoro. Consulta Proprietà cluster.

Scalabilità automatica e Spark Structured Streaming

La scalabilità automatica non è compatibile con Streaming strutturato di Spark dal Al momento, lo streaming strutturato Spark non supporta l'allocazione dinamica (vedi SPARK-24815: lo streaming strutturato deve supportare l'allocazione dinamica).

Controllo della scalabilità automatica tramite partizionamento e parallelismo

Sebbene il parallelismo sia in genere impostato o determinato dalle risorse del cluster (ad esempio, il numero di blocchi HDFS viene controllato dal numero di attività), con la scalabilità automatica si applica il contrario: la scalabilità automatica imposta il numero di worker in base al parallelismo del job. Di seguito sono riportate alcune linee guida per aiutarti a impostare il parallelismo dei lavori:

• Sebbene Dataproc imposti il numero predefinito di attività di riduzione MapReduce in base alle dimensioni iniziali del cluster, puoi impostare mapreduce.job.reduces per aumentare il parallelismo della fase di riduzione.
• Il parallelismo di Spark SQL e Dataframe è determinato da spark.sql.shuffle.partitions, che ha un valore predefinito di 200.
• Per impostazione predefinita, le funzioni RDD di Spark utilizzano spark.default.parallelism, che viene impostato sul numero di core sui nodi worker all'avvio del job. Tuttavia, tutte le funzioni RDD che creano rimescolamenti accettano un parametro per il numero di partizioni, che sostituisce spark.default.parallelism.

Devi assicurarti che i dati siano partizionati in modo uniforme. In caso di disallineamento significativo dei tasti, una o più attività possono richiedere molto più tempo rispetto ad altre, con conseguente con un utilizzo ridotto.

Impostazioni predefinite della proprietà Spark/Hadoop con scalabilità automatica

I cluster con scalabilità automatica hanno valori predefiniti per le proprietà del cluster che aiutano a evitare l'errore del job quando i worker principali vengono rimossi o i worker secondari vengono prelevati. Puoi eseguire l'override a questi valori predefiniti quando crei un cluster con scalabilità automatica (vedi Proprietà del cluster).

Per impostazione predefinita, il numero massimo di nuovi tentativi per attività, master dell'applicazione e fasi viene aumentato:

yarn:yarn.resourcemanager.am.max-attempts=10
mapred:mapreduce.map.maxattempts=10
mapred:mapreduce.reduce.maxattempts=10
spark:spark.task.maxFailures=10
spark:spark.stage.maxConsecutiveAttempts=10

Per impostazione predefinita, reimposta i contatori di ripetizione (utile per i job Spark Streaming di lunga durata):

spark:spark.yarn.am.attemptFailuresValidityInterval=1h
spark:spark.yarn.executor.failuresValidityInterval=1h

L'impostazione predefinita è l'avvio lento di Spark. meccanismo di allocazione dinamica a partire da una dimensione ridotta:

spark:spark.executor.instances=2

Domande frequenti

La scalabilità automatica può essere attivata nei cluster ad alta disponibilità e nei cluster a nodo singolo?

La scalabilità automatica può essere abilitata cluster ad alta disponibilità, ma non su cluster a nodo singolo (I cluster a nodo singolo non supportano il ridimensionamento).

È possibile ridimensionare manualmente un cluster con scalabilità automatica?

Sì. Puoi decidere di ridimensionare manualmente un cluster come misura temporanea quando ottimizzi un criterio di scalabilità automatica. Tuttavia, queste modifiche avranno solo un e Scalabilità automatica ridimensiona il cluster.

Anziché ridimensionare manualmente un cluster con scalabilità automatica, prendi in considerazione i seguenti aspetti:

Aggiornamento del criterio di scalabilità automatica. Qualsiasi modifica apportata al criterio di scalabilità automatica influirà su tutti i cluster che attualmente lo utilizzano (consulta Utilizzo dei criteri multi-cluster).

Scollegamento del criterio e scalabilità manuale con la dimensione che preferisci.

Assistenza per Dataproc.

Qual è la differenza tra Dataproc e la scalabilità automatica di Dataflow?

Consulta Scalabilità automatica orizzontale di Dataflow e la scalabilità automatica verticale di Dataflow.

Il team di sviluppo di Dataproc può reimpostare lo stato di un cluster da ERROR a RUNNING?

In generale, no. Questa operazione richiede un intervento manuale per verificare se è sicuro semplicemente reimpostando lo stato del cluster, che spesso non può essere reimpostato altre procedure manuali, ad esempio il riavvio del Namenode di HDFS.

Dataproc imposta lo stato di un cluster su ERROR quando ma non riesce a determinare lo stato di un cluster dopo un'operazione non riuscita. Cluster in ERROR non prevede la scalabilità automatica o non esegue job. Le cause più comuni sono:

1. Errori restituiti dall'API Compute Engine, spesso durante le interruzioni di Compute Engine.

2. HDFS entra in uno stato danneggiato a causa di bug nel ritiro di HDFS.

3. Errori dell'API Dataproc Control, ad esempio "Lease dell'attività scaduto"

Elimina e ricrea i cluster il cui stato è ERROR.

Quando la scalabilità automatica annulla un'operazione di scale down?

L'immagine seguente è un'illustrazione che mostra quando la scalabilità automatica annullare un'operazione di scale down (vedi anche Come funziona la scalabilità automatica).

Note:

• Nel cluster è abilitata la scalabilità automatica solo in base alle metriche di memoria YARN (impostazione predefinita).
• T1-T9 rappresentano gli intervalli di attesa quando il gestore della scalabilità automatica valuta il numero di worker (la temporizzazione degli eventi è stata semplificata).
• Le barre in pila rappresentano il numero di attività, dismesse e dismesse worker YARN del cluster.
• Il numero consigliato di worker dell'autoscaler (linea nera) si basa sulle metriche della memoria YARN, sul conteggio dei worker attivi YARN e sulle impostazioni dei criteri di scalabilità automatica (consulta Come funziona la scalabilità automatica).
• L'area di sfondo rossa indica il periodo in cui è in esecuzione l'operazione di riduzione.
• L'area di sfondo gialla indica il periodo in cui l'operazione di riduzione viene annullata.
• L'area di sfondo verde indica il periodo dell'operazione di scalabilità.

Le seguenti operazioni vengono eseguite nei seguenti momenti:

• T1: il gestore della scalabilità automatica avvia un'operazione di scaledown di rimozione controllata fare lo scale down di circa la metà dei worker attuali del cluster.

• T2: il gestore della scalabilità automatica continua a monitorare le metriche del cluster. Non cambia suggerimento di scale down e l'operazione di scale down continua. Alcuni lavoratori sono stati dismessi, altri stanno dismesso (Dataproc eliminerà i lavoratori dismessi).

• T3: Il gestore della scalabilità automatica calcola che il numero di worker può fare lo scale down ulteriore, probabilmente a causa della disponibilità di memoria YARN aggiuntiva. Tuttavia, poiché il numero di worker attivi più la variazione consigliata nel numero di worker non è uguale o superiore al numero di worker attivi più quelli in fase di ritiro, i criteri per l'annullamento del ridimensionamento non sono soddisfatti e il ridimensionamento automatico non annulla l'operazione di ridimensionamento.

• T4: YARN segnala un aumento della memoria in attesa. Tuttavia, il gestore della scalabilità automatica non modifica il numero di worker consigliato. Come nel T3, lo scaledown i criteri di annullamento non vengono soddisfatti non annulla l'operazione di scale down.

• T5: YARN in attesa dell'aumento della memoria e modifica del numero di worker consigliati dal gestore della scalabilità automatica. Tuttavia, poiché il numero di worker attivi più la modifica consigliata nel numero di worker è minore del di worker attivi più quelli in dismissione, i criteri di annullamento rimangono non soddisfatto e l'operazione di scale down non viene annullata.

• T6: la memoria YARN in attesa aumenta ulteriormente. Il numero di worker attivi più la variazione del numero di worker consigliati dall'autoscaler ora è maggiore del numero di worker attivi più quelli in fase di ritiro. Criteri di annullamento vengono soddisfatti e il gestore della scalabilità automatica annulla l'operazione di scale down.

• T7: il gestore della scalabilità automatica è in attesa del completamento dell'annullamento dell'operazione di riduzione. Il gestore della scalabilità automatica non valuta e consiglia una modifica del numero di worker durante questo intervallo.

• T8: l'annullamento dell'operazione di riduzione viene completato. Disattivazione dei worker vengono aggiunti al cluster e diventano attivi. Il gestore della scalabilità automatica rileva l'annullamento dell'operazione di scaledown e attende la successiva di valutazione (T9) per calcolare il numero consigliato di worker.

• T9: non ci sono operazioni attive all'orario T9. In base alle norme del gestore della scalabilità automatica e YARN, il gestore della scalabilità automatica consiglia un'operazione di scale up.