Scalabilità automatica dei cluster Dataproc

Obiettivo: capire come configurare e utilizzare Dataproc della scalabilità automatica per scalare in modo automatico e dinamico il numero di VM worker Cluster Dataproc per soddisfare le esigenze dei carichi di lavoro.

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. Avviato dall'utente Scalabilità dei cluster risolve parzialmente questa sfida, ma richiede il monitoraggio dell'utilizzo del cluster e interventi manuali.

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. Un Autoscaling Policy è una configurazione riutilizzabile che descrive in che modo devono essere scalati 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.

• La scalabilità automatica di Dataproc esegue il ridimensionamento dei nodi worker del cluster, non dei nodi master del cluster.
• La scalabilità automatica di Dataproc supporta la scalabilità (scalabilità del numero di nodi) e non la scalabilità verticale. (scalabilità dei tipi di macchine).

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 indicatore 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 datanode copiano i blocchi HDFS in altri datanode 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 di YARN-9088. YARN registra in modo errato le metriche del cluster quando vengono utilizzate le etichette dei nodi.

• Spark Structured Streaming: la scalabilità automatica non supporta Spark Structured Streaming (consulta Scalabilità automatica e Spark Structured Streaming).

• 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. Dal giorno per creare un nuovo cluster basta ridimensionarne uno, valuta la possibilità di eliminare i cluster inattivi e ricreandoli. Sono supportati dai seguenti strumenti "temporaneo" modello:

  Utilizza i workflow di Dataproc per pianificare un insieme di job su un cluster dedicato ed elimina 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. Di conseguenza, un job che richiede molto tempo ritarderà la scalabilità automatica delle risorse per i job più piccoli in esecuzione sul cluster fino al termine del job che richiede molto tempo. Per evitare questo risultato, raggruppa i job più piccoli di dimensioni simili su isola ogni job di lunga durata su un cluster a parte.

Abilitazione della scalabilità automatica

Per abilitare la scalabilità automatica su un cluster:

1. Crea un criterio di scalabilità automatica.

2. In uno dei seguenti modi:

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

Crea un criterio di scalabilità automatica

Comando g-cloud

Puoi utilizzare il comando gcloud dataproc autoscaling-policies import per creare un criterio di scalabilità automatica. Legge un locale YAML che definisce un criterio di scalabilità automatica. Il formato e i contenuti del file deve corrispondere agli oggetti di configurazione e ai campi definiti autoscalingPolicies l'API REST.

Il seguente esempio YAML definisce un criterio che specifica tutti i campi. Fornisce inoltre i valori minInstances e maxInstances per i worker principali, il valore maxInstances per i worker secondari (prerilasciabili) e specifica un cooldownPeriod di 4 minuti (il valore predefinito è di 2 minuti). workerConfig configura i worker principali. Nella in questo esempio, minInstances e maxInstances sono Impostato sullo stesso valore per evitare la scalabilità dei worker principali.

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

Ecco un altro esempio YAML che specifica tutti i campi facoltativi e obbligatori dei criteri di scalabilità automatica.

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

Esegui questo comando gcloud da un terminale locale o in Cloud Shell per creare e il criterio di scalabilità automatica. Specifica un nome per il criterio. Questo nome diventerà il criterio id, che potrai usare in un secondo momento gcloud per fare riferimento al criterio. Utilizza il flag --source per specificare il percorso file locale e il nome file del file YAML del criterio di scalabilità automatica da importare.

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

API REST

Crea un criterio di scalabilità automatica definendo un AutoscalingPolicy all'interno di una richiesta autoscalingPolicies.create.

Console

Per creare un criterio di scalabilità automatica, seleziona CREA CRITERIO da Dataproc Criteri di scalabilità automatica utilizzando la console Google Cloud. Il giorno Crea criterio puoi selezionare un riquadro dei consigli relativi alle norme per compilare campi dei criteri di scalabilità automatica per un tipo di job o un obiettivo di scalabilità specifici.

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.

Comando g-cloud

Esegui questo comando gcloud da un terminale locale o in Cloud Shell per creare con scalabilità automatica. Fornisci un nome per il cluster e utilizza il flag --autoscaling-policy per specificare policy id (il nome del criterio specificato quando hai creato il criterio) o il criterio resource URI (resource name) (consulta i campi AutoscalingPolicy id e name).

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

Il nome completo della risorsa (o URI) di un criterio puoi ottenerlo eseguendo questo gcloud :

gcloud dataproc autoscaling-policies \
    describe policy-id \
    --region=region

API REST

Crea un cluster con la scalabilità automatica includendo un AutoscalingConfig all'interno di una richiesta clusters.create.

Per visualizzare AutoscalingConfig.policyUri (nome della risorsa criterio), esegui il seguente comando gcloud :

gcloud dataproc autoscaling-policies \
    describe policy-name \
    --region=region

Console

Puoi selezionare un criterio di scalabilità automatica esistente da applicare a un nuovo cluster dalla sezione Criterio di scalabilità automatica del riquadro Configura cluster nella pagina Dataproc Crea un cluster della console Google Cloud.

Abilita la scalabilità automatica su un cluster esistente

Dopo aver creato un criterio di scalabilità automatica, puoi abilitare il criterio su un cluster esistente regione.

Comando g-cloud

Esegui il seguente comando gcloud da un terminale locale o in Cloud Shell per abilitare un criterio di scalabilità automatica in un cluster esistente. Fornisci il nome del cluster e utilizza il flag --autoscaling-policy per specificare policy id (il nome del criterio che hai specificato quando hai ha creato il criterio) o le norme resource URI (resource name) (consulta Criteri di scalabilità automatica id e name campi).

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

Il nome completo della risorsa (o URI) di un criterio puoi ottenerlo eseguendo questo gcloud :

gcloud dataproc autoscaling-policies \
    describe policy-id \
    --region=region

API REST

Per attivare un criterio di scalabilità automatica su un cluster esistente, imposta AutoscalingConfig.policyUri del criterio in updateMask di una richiesta clusters.patch.

Per visualizzare AutoscalingConfig.policyUri (nome risorsa del criterio), esegui il seguente gcloud :

gcloud dataproc gcloud dataproc autoscaling-policies \
    describe policy-name \
    --region=region

Console

Al momento, l'abilitazione di un criterio di scalabilità automatica su un cluster esistente non è supportata nella console Google Cloud.

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 e schemi ricorrenti.

• Puoi aggiornare un criterio utilizzato da più cluster. Gli aggiornamenti influiscono immediatamente sul comportamento della scalabilità automatica per tutti i cluster che utilizzano il criterio (vedi autoscalingPolicies.update). Se non vuoi che un aggiornamento del criterio venga applicato a un cluster che lo utilizza, disattiva la scalabilità automatica sul cluster prima di aggiornare il criterio.

Comando g-cloud

Esegui questo comando gcloud da un terminale locale o in da Cloud Shell a di disabilitare la scalabilità automatica su un cluster.

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

API REST

Per disattivare la scalabilità automatica su un cluster, imposta AutoscalingConfig.policyUri sulla stringa vuota e imposta update_mask=config.autoscaling_config.policy_uri in una richiesta clusters.patch.

Console

Al momento, la disattivazione della scalabilità automatica su un cluster non è supportata nella console Google Cloud.

• Un criterio in uso 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 al termine di ogni periodo di "attesa" per determinare se eseguire lo scaling del cluster e, in caso affermativo, la portata dell'aggiornamento.

1. Valore della metrica delle risorse in attesa YARN (Memoria in attesa o Core in attesa) determina se fare lo scale up o lo scale down. Un valore maggiore di 0 indica che i job YARN sono in attesa di risorse e potrebbe essere necessario fare lo scale up. 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 è > 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 è pari a 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 attivo 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 \]

   Nota: i worker secondari, inclusi quelli la cui creazione è pianificata, sono inclusi nel numero di worker corrente.

2. Data la variazione stimata necessaria al numero di worker, la scalabilità automatica utilizza un valore scaleUpFactor o scaleDownFactor per calcolare la variazione effettiva del 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
   

   Uno scaleUpFactor o scaledownFactor di 1,0 indica che la scalabilità automatica scalerà la risorsa in attesa/disponibile è 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à la scalabilità del cluster. Una piccola frazione indica che la scalabilità automatica deve scalare anche se Δworkers è un numero 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 è di scalabilità al cluster per la scalabilità punto.

5. Le richieste di scale down della scalabilità automatica verranno annullate sui cluster creati con un'immagine versioni successive alla 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 ("worker attivi") più la variazione del numero totale di worker consigliati dal gestore della scalabilità automatica (Δworkers) è uguale o superiore a DECOMMISSIONING worker YARN ("worker di disattivazione"), come mostrato nella formula seguente:

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

      La formula di annullamento dell'operazione di scale down prendere in considerazione il numero di worker attivi, non current_worker_count (utilizzato nell'esperienza precedente formule di scalabilità automatica).

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

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'utilizzo di worker secondari ti consente di scalare in modo efficiente le risorse di calcolo senza dover eseguire il provisioning dello spazio di archiviazione, con conseguente aumento delle funzionalità 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

Devi apportare alcune modifiche al comando di creazione del cluster:

1. Imposta --num-workers=10 in modo che corrisponda alle dimensioni del gruppo di worker principale del criterio di scalabilità automatica.
2. Imposta --secondary-worker-type=non-preemptible per configurare i worker secondari come 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 scalabilità automatica per:

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

per evitare interruzioni dei job in esecuzione rispetto allo scale down 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.

Scelta di un timeout per la rimozione controllata

La scalabilità automatica supporta Ritiro gestito YARN durante la rimozione dei nodi da un cluster. Il ritiro graduale consente alle applicazioni di completare l'ordinamento casuale dei dati tra le fasi per evitare di reimpostare l'avanzamento del job. La Il timeout per il ritiro gestito automaticamente fornito in un criterio di scalabilità automatica è limite superiore del tempo di attesa di YARN per l'esecuzione delle applicazioni ( che erano in esecuzione quando è iniziato il ritiro) prima di rimuovere i nodi.

Nota: la scalabilità automatica non farà lo scale up durante la rimozione controllata. Aspetta per il completamento di un'operazione di aggiornamento, tra cui rimozione controllata, continua ad attendere un periodo di assestamento (ad esempio, 2 min), poi emette un altro di aggiornamento.

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 gestito automaticamente su un valore maggiore del job più lungo che verrà elaborato dal cluster. Ad esempio, se prevedi che la tua attività un job affinché venga eseguito per 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 per la loro transizione temporanea per evitare di bloccare la rimozione controllata.

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 di risorsa YARN in attesa che causa l'aggiunta di nodi.

Ad esempio, se ci sono 100 container in attesa che richiedono 512 MB ciascuno, ci sono 50 GB di memoria YARN in attesa. Se scaleUpFactor è 0.5, il gestore della scalabilità automatica aggiungerà nodi sufficienti per aggiungere 25 GB di memoria YARN. Analogamente, 0.1, il gestore della scalabilità automatica 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 fisso di esecuzioni e i job Tez, utilizza 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 riduce un cluster. Specifica un numero compreso tra 0.0 e 1.0 per impostare il valore frazionario di 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 maggiore stabilità, imposta un valore scaleDownFactor più basso per una riduzione della velocità di scale down.

Imposta questo valore su 0.0 per impedire lo scale down del cluster, ad esempio quando mediante cluster temporanei o con job singolo.

Nota: se imposti sia scaleUpFactor sia scaleDownFactor su 0, il tuo cluster non verrà scalato.

Impostazione 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 alle quali il gestore della scalabilità automatica fare lo scale up o fare lo scale down del cluster: l'aumento minimo del valore frazionario o diminuire le dimensioni del cluster necessarie per inviare richieste di scale up o scale down.

Esempi: il gestore della scalabilità automatica non invierà 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. Analogamente, se scaleDownMinWorkerFraction è 0.05, il gestore della scalabilità non emetterà una richiesta di aggiornamento per rimuovere i nodi da un cluster di 100 nodi, a meno che non debbano essere rimossi almeno 5 nodi.

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

Impostando una posizione superiore scaleDownMinWorkerFractionthresholds su cluster di grandi dimensioni (> 100 nodi) per evitare piccole operazioni di scalabilità non necessarie, 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 usarlo per limitare la frequenza delle modifiche alla dimensione del cluster del gestore della scalabilità automatica.

Il valore minimo e predefinito di cooldownPeriod è di due minuti. Se in un criterio è impostato un valore cooldownPeriod più breve, le modifiche al carico di lavoro influiranno più rapidamente sulle dimensioni del cluster, ma i cluster potrebbero essere scalati inutilmente. È consigliabile impostare il valore scaleUpMinWorkerFraction e scaleDownMinWorkerFraction su un valore diverso da zero se utilizzi un valore cooldownPeriod più corto. In questo modo, il cluster viene scalato verso l'alto o verso il basso solo quando la variazione nell'utilizzo delle risorse è sufficiente a giustificare 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. Sperimenta diversi periodi di attesa per trovare il valore più adatto a per il tuo carico di lavoro.

Limiti del numero di worker e pesi del 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 bilanciamento target 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.

Nota: se minInstances o maxInstances vengono aggiornati in un criterio di scalabilità automatica per fare in modo che le dimensioni di un cluster superino i limiti, il gestore della scalabilità automatica lo scale up o lo scale down ai nuovi limiti. Il gestore della scalabilità automatica eviterà e lo scale up del cluster oltre i limiti. Consulta https://issuetracker.google.com/160640545.

Attenzione: se imposti minInstances = maxInstances, il tuo cluster non verrà scalato.

Abilita la scalabilità automatica in base ai 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 dei nodi nel cluster
• comprendi perché la scalabilità automatica ha scalato o meno 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 consigli per la scalabilità, le metriche utilizzate per le decisioni di scalabilità, le dimensioni del cluster originale e le dimensioni del nuovo cluster di destinazione.

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

Puoi configurare la scalabilità automatica per qualsiasi di esecuzione dei pod.

Nota: la "risorsa" YARN qui può essere la memoria YARN o il nucleo YARN.

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

1. Allocated resource si riferisce alla risorsa YARN totale utilizzata dai container in esecuzione nell'intero cluster. Se sono in esecuzione sei contenitori che possono utilizzare fino a 1 unità di risorsa, sono disponibili sei risorse allocate.

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 disponibili (non utilizzate) nel cluster, la scalabilità automatica potrebbe rimuovere 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 contenitori in attesa, la scalabilità automatica potrebbe aggiungere worker al cluster.

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 attività di mappatura e riduzione come contenitore YARN separato. Quando inizia un job, MapReduce invia richieste di contenitori per ogni attività di mappatura, con un conseguente picco elevato della 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. L'aggiunta di worker al cluster richiede almeno 1,5 minuti, quindi assicurati che sia presente un lavoro in sospeso sufficiente per utilizzare il nuovo worker per diversi 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 oltre 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. Questa operazione alleggerisce la memoria in attesa (meno picchi di memoria in sospeso). 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 estrai i set di dati dalla cache 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 la propria versione allocazione dinamica che utilizza indicatori specifici dei flussi di dati per aggiungere e rimuovere gli esecutori, imposta spark.streaming.dynamicAllocation.enabled=true e disabilita 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. Invece, per rimuovere in sicurezza i lavoratori con scalabilità automatica, configura i checkpoint per la tolleranza di errore.

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 il tuo job. Vedi Proprietà del 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 job:

• Mentre Dataproc imposta il numero predefinito di riduzione MapReduce basate sulle dimensioni iniziali dei 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 per impostazione predefinita è 200.
• Il valore predefinito delle funzioni RDD di Spark è spark.default.parallelism, impostato su il numero di core sui nodi worker all'avvio del job. Tuttavia, tutti gli RDD per le funzioni che creano shuffling prendono una parametro per il numero di partizioni, che sostituisce spark.default.parallelism.

Devi assicurarti che i dati siano partizionati in modo uniforme. Se lo skew delle chiavi è significativo, una o più attività potrebbero richiedere molto più tempo rispetto ad altre, con un conseguente utilizzo ridotto.

Impostazioni predefinite delle proprietà Spark/Hadoop per la scalabilità automatica

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

Per aumentare la stabilità dei job, puoi creare un cluster con worker secondari non prerilasciabili. Per un esempio, consulta Utilizzare i prerilasciabili in un cluster.

Per impostazione predefinita, il numero massimo di nuovi tentativi per attività, master di applicazioni 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

Per impostazione predefinita, il meccanismo di allocazione dinamica con avvio lento di Spark inizia da una dimensione ridotta:

spark:spark.executor.instances=2

Nota: anche se potresti notare errori di recupero dello shuffling quando i worker principali vengono rimossi o i worker secondari vengono prerilasciati, non devi aumentare i nuovi tentativi di recupero dello shuffling in alto i relativi valori predefiniti (spark.shuffle.io.maxRetries=3; spark.shuffle.io.retryWaitMs=5000 (5 secondi) per un valore totale massimo dei nuovi tentativi di 10 secondi. Il motivo: i file in shuffling vengono persi, quindi i nuovi tentativi richiedono solo più tempo.

Domande frequenti

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

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

Puoi 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).

Scollegare il criterio e ridimensionare manualmente il cluster in base alle dimensioni preferite.

Ricevere assistenza per Dataproc.

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

Consulta Scalabilità automatica orizzontale di Dataflow e Scalabilità automatica verticale di Dataflow Prime.

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 reimpostare lo stato del cluster e spesso un cluster non può essere reimpostato senza altri passaggi manuali, come 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 Compute Engine o in caso di interruzione del servizio.

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

3. Errori dell'API Dataproc Control come "Leasing attività scaduto"

Elimina e ricrea i cluster il cui stato è ERROR.

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

La seguente illustrazione mostra quando la scalabilità automatica annullerà un'operazione di riduzione (consulta anche Come funziona la scalabilità automatica).

Note:

• La scalabilità automatica del cluster è attivata solo in base alle metriche della memoria YARN (impostazione predefinita).
• I livelli T1-T9 rappresentano gli intervalli di assestamento quando il gestore della scalabilità automatica valuta il numero di worker (la tempistica degli eventi è stata semplificata).
• Le barre in pila rappresentano i conteggi dei worker YARN del cluster attivi, in fase di ritiro e ritirati.
• 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 si verificano ai seguenti orari:

• 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: l'autoscalatore calcola che il numero di worker può essere ridotto ulteriormente, possibly due to additional YARN memory becoming available. 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 gestore della scalabilità automatica non annulla l'operazione di ridimensionamento.

• T4: YARN registra un aumento della memoria in attesa. Tuttavia, il gestore della scalabilità automatica non cambia il suggerimento sul conteggio dei worker. Come nel T3, lo scaledown i criteri di annullamento non vengono soddisfatti non annulla l'operazione di scale down.

• T5: la memoria in attesa di YARN aumenta e aumenta la variazione del numero di worker consigliati dall'agente di 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 dell'annullamento dell'operazione di scale down per completato. Il gestore della scalabilità automatica non valuta e consiglia una modifica del numero di worker durante questo intervallo.

• T8: l'annullamento dell'operazione di scale down viene completata. I worker di ritiro 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.