Scalabilità automatica dei cluster

Che cos'è la scalabilità automatica

Stimare il numero "giusto" di worker (nodi) del cluster per un workload è difficile e una singola dimensione del cluster per un'intera pipeline spesso non è l'ideale. La scalabilità dei cluster avviata dall'utente affronta parzialmente questa sfida, ma richiede il monitoraggio dell'utilizzo del cluster e l'intervento manuale.

L'API AutoscalingPolicies di Dataproc fornisce un meccanismo per automatizzare la gestione delle risorse cluster e consente la scalabilità automatica delle VM worker del cluster. Un Autoscaling Policy è una configurazione riutilizzabile che descrive come i worker del cluster che utilizzano la policy di scalabilità automatica devono scalare. Definisce i limiti, la frequenza e l'aggressività della scalabilità per fornire un controllo granulare sulle risorse del cluster per tutta la sua durata.

Quando utilizzare la scalabilità automatica

Utilizza la scalabilità automatica:

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

su cluster che elaborano molti job

per fare lo scale up dei cluster per 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 è destinata alla scalabilità di 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 primari. Il numero di worker primari deve essere sufficiente per ospitare tutti i dati HDFS.
    3. La disattivazione dei DataNode 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, questo processo può richiedere 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 segnala 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 fare lo scale down di un cluster alle dimensioni minime quando il cluster è inattivo. Poiché la creazione di un nuovo cluster è veloce quanto il ridimensionamento, valuta la possibilità di eliminare i cluster inattivi e ricrearli. I seguenti strumenti supportano questo modello "temporaneo":

    Utilizza Workflows di Managed Service for Apache Spark per pianificare un insieme di job su un cluster dedicato, quindi elimina il cluster al termine dei job. Per un'orchestrazione più avanzata, utilizza Cloud Composer, basato su Apache Airflow.

    Per i cluster che elaborano query ad hoc o workload pianificati esternamente, utilizza l'eliminazione pianificata del cluster per eliminare il cluster dopo un periodo di inattività o una durata specificati o in un momento specifico.

  • Workload di dimensioni diverse: quando su un cluster vengono eseguiti job di piccole e grandi dimensioni, lo scale down rimozione controllata attende il completamento dei job di grandi dimensioni. Di conseguenza, un job a lunga esecuzione ritarderà la scalabilità automatica delle risorse per i job più piccoli in esecuzione sul cluster fino al termine del job a lunga esecuzione. Per evitare questo risultato, raggruppa i job più piccoli di dimensioni simili su un cluster e isola ogni job di lunga durata su un cluster separato.

Abilita scalabilità automatica

Per abilitare la scalabilità automatica su un cluster:

  1. Crea una policy di scalabilità automatica.

  2. Uno dei seguenti:

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

Crea una policy di scalabilità automatica

ConsoleGoogle Cloud

Per creare una policy di scalabilità automatica, seleziona CREA POLICY dalla pagina Policy di scalabilità automatica di Managed Service for Apache Spark utilizzando la console Google Cloud . Nella pagina Crea policy, puoi selezionare un riquadro dei suggerimenti per le policy per compilare i campi della policy di scalabilità automatica per un tipo di prestazione o un obiettivo di scalabilità specifico.

gcloud CLI

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

L'esempio YAML seguente definisce una policy per i cluster standard di Managed Service for Apache Spark, con tutti i campi obbligatori. Inoltre, fornisce i valori minInstances e maxInstances per i worker primari, il valore maxInstances per i worker secondari e specifica un cooldownPeriod di 4 minuti (il valore predefinito è 2 minuti). workerConfig configura i worker primari. In questo esempio, minInstances e maxInstances sono impostati sullo stesso valore per evitare la scalabilità dei worker primari. 

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

L'esempio YAML seguente definisce una policy per i cluster standard di Managed Service for Apache Spark, con tutti i campi obbligatori e facoltativi della policy di scalabilità automatica.

clusterType: STANDARD
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

L'esempio YAML seguente definisce una policy per i cluster con scalabilità zero.

Per i cluster con scalabilità zero, non includere workerConfig.

clusterType: ZERO_SCALE
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 il seguente comando gcloud da un terminale locale o in Cloud Shell per creare la policy di scalabilità automatica. Specifica un nome per la policy. Questo nome diventerà la id della policy, che potrai utilizzare nei comandi gcloud successivi per fare riferimento alla policy. Utilizza il flag --source per specificare il percorso locale e il nome file del file YAML della policy di scalabilità automatica da importare. 

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

API REST

Crea una policy di scalabilità automatica definendo un AutoscalingPolicy come parte di una richiesta autoscalingPolicies.create.

Crea un cluster con scalabilità automatica

Dopo aver creato una policy di scalabilità automatica, crea un cluster che utilizzerà la policy di scalabilità automatica. Il cluster deve trovarsi nella stessa regione della policy di scalabilità automatica.

ConsoleGoogle Cloud

Per applicare una policy di scalabilità automatica a un nuovo cluster:

  1. Apri la pagina Crea cluster di Managed Service for Apache Spark.
  2. Nella sezione Definisci il cluster, in Scegli la modalità di scalabilità dei worker, seleziona Scalabilità automatica, quindi seleziona la policy nel campo Policy.

gcloud CLI

Esegui il seguente comando gcloud da un terminale locale o in Cloud Shell per creare un cluster con scalabilità automatica. Specifica un nome per il cluster e utilizza il flag --autoscaling-policy per indicare policy ID (il nome della policy specificato quando hai creato la policy) o la policy 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

API REST

Crea un cluster con scalabilità automatica includendo un AutoscalingConfig come parte di una richiesta clusters.create.

Abilita la scalabilità automatica su un cluster esistente

Dopo aver creato una policy di scalabilità automatica, puoi attivarla su un cluster esistente nella stessa regione.

ConsoleGoogle Cloud

L'attivazione di una policy di scalabilità automatica su un cluster esistente non è supportata nella console Google Cloud .

gcloud CLI

Esegui il seguente comando gcloud da un terminale locale o in Cloud Shell per abilitare una policy di scalabilità automatica su un cluster esistente. Specifica il nome del cluster e utilizza il flag --autoscaling-policy per indicare policy ID (il nome della policy specificato quando hai creato la policy) o la policy resource URI (resource name) (consulta i campi AutoscalingPolicy id e name).

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

API REST

Per attivare una policy di scalabilità automatica su un cluster esistente, imposta AutoscalingConfig.policyUri della policy nel updateMask di una richiesta clusters.patch.

Utilizzo delle policy multi-cluster

  • Una policy di scalabilità automatica definisce il comportamento di scalabilità che può essere applicato a più cluster. Una policy di scalabilità automatica viene applicata al meglio su più cluster quando questi condividono workload simili o eseguono job con pattern di utilizzo delle risorse simili.

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

ConsoleGoogle Cloud

La disattivazione della scalabilità automatica su un cluster non è supportata nella console Google Cloud .

gcloud CLI

Esegui il seguente comando gcloud da un terminale locale o in Cloud Shell per disattivare 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.

Come funziona la scalabilità automatica

La scalabilità automatica controlla le metriche YARN Hadoop del cluster al termine di ogni periodo di "cooldown" per determinare se scalare il cluster e, in caso affermativo, l'entità dell'aggiornamento.

  1. Il valore della metrica delle risorse YARN in attesa (Memoria in attesa o Core in attesa) determina se eseguire lo scale up o lo scale down. Un valore maggiore di 0 indica che i job YARN sono in attesa di risorse e che potrebbe essere necessario fare lo scale up. Un valore 0 indica che YARN dispone di risorse sufficienti, pertanto potrebbe non essere necessario fare lo scale down o apportare 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 è 0:

    $estimated\_worker\_count =$

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

    Per impostazione predefinita, a partire dall'immagine 2.2 di Managed Service for Apache Spark, il gestore della scalabilità automatica monitora la memoria e i core YARN in modo cheestimated_worker_count venga valutato separatamente per la memoria e i core e venga selezionato il numero di worker più grande risultante. Per le versioni immagine precedenti, il gestore della scalabilità automatica monitora solo la memoria YARN, a meno che non attivi la scalabilità automatica basata sui core per monitorare sia la memoria YARN sia i core.

    $estimated\_worker\_count =$

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

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

  2. In base alla variazione stimata necessaria per il numero di worker, la scalabilità automatica utilizza un scaleUpFactor o un 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

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

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

    IF (Δworkers >  scaleUpMinWorkerFraction * current_worker_count) then scale up
    OR 
    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 primari e secondari) per determinare come suddividere il numero di worker tra i gruppi di istanza worker primari e secondari. Il risultato di questi calcoli è la modifica finale della scalabilità automatica del cluster per il periodo di scalabilità.

  5. Le richieste di scale down della scalabilità automatica verranno annullate sui cluster creati con le versioni immagine 2.0.57+, 2.1.5+ e successive se:

    1. è in corso uno scale down con un valore di timeout di rimozione controllata 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 in fase di ritiro"), come mostrato nella seguente formula:

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

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

Suggerimenti per la configurazione della scalabilità automatica

Questa sezione contiene suggerimenti per aiutarti a configurare la scalabilità automatica.

Evita di scalare i worker primari

I worker primari eseguono i datanode HDFS, mentre i worker secondari sono solo di computing. L'utilizzo di worker secondari consente di scalare in modo efficiente le risorse di computing senza la necessità di eseguire il provisioning dello spazio di archiviazione, con conseguenti funzionalità di scalabilità più rapide. I NameNode HDFS possono avere più race condition che causano il danneggiamento di HDFS, per cui il ritiro si blocca a tempo indeterminato. Per evitare questo problema, evita di scalare i worker primari. Ad esempio: none workerConfig: minInstances: 10 maxInstances: 10 secondaryWorkerConfig: minInstances: 0 maxInstances: 100

Per creare il cluster, devi apportare alcune modifiche al comando:

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

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

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

consentono di fare lo scale down del cluster più velocemente mentre i job sono in esecuzione

per evitare interruzioni dei job in esecuzione dovute allo scale down del cluster

ridurre al minimo l'interruzione dei job in esecuzione a causa del prerilascio dei worker secondari preemptible

Con EFM attivato, il timeout della dismissione controllata di una policy di scalabilità automatica deve essere impostato su 0s. La policy di scalabilità automatica deve scalare automaticamente solo i worker secondari.

Scegli un timeout per rimozione controllata

La scalabilità automatica supporta la dismissione controllata di YARN durante la rimozione dei nodi da un cluster. La dismissione controllata consente alle applicazioni di completare lo shuffling dei dati tra le fasi per evitare di ritardare l'avanzamento del job. Il timeout di dismissione controllata fornito in una policy di scalabilità automatica è il limite superiore della durata di attesa di YARN per le applicazioni in esecuzione (applicazioni in esecuzione all'inizio della dismissione) prima della rimozione dei nodi.

Se un processo non viene completato entro il periodo di timeout rimozione controllata specificato, il nodo worker viene arrestato forzatamente, il che potrebbe causare la perdita di dati o l'interruzione del servizio. Per evitare questa possibilità, imposta il timeout della dismissione controllata su un valore superiore al job più lungo che verrà elaborato dal cluster. Ad esempio, se prevedi che il job più lungo 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 ai propri cluster temporanei per evitare di bloccare la rimozione controllata.

Configurazione di scaleUpFactor in corso…

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

Ad esempio, se sono presenti 100 container in attesa che richiedono 512 MB ciascuno, la memoria YARN in attesa è di 50 GB. Se scaleUpFactor è 0.5, il gestore della scalabilità automatica aggiungerà un numero di nodi sufficiente per aggiungere 25 GB di memoria YARN. Analogamente, se è 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 i job Spark con l'allocazione dinamica abilitata. Per i job Spark con un numero fisso di esecutori e i job Tez, utilizza 1.0. Un valore scaleUpFactor pari a 1.0 indica che la scalabilità automatica verrà eseguita in modo che la risorsa in attesa o disponibile sia pari a 0 (utilizzo perfetto).

Configurazione di scaleDownFactor in corso…

scaleDownFactor controlla l'aggressività con cui il gestore della scalabilità automatica esegue lo scale down di un cluster. 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 fare lo scale up e lo scale down di frequente. A causa della rimozione controllata, le operazioni di scale down sono notevolmente più lente di quelle di scale up. L'impostazione scaleDownFactor=1.0 imposta una frequenza di scale down aggressiva, che riduce al minimo il numero di operazioni di scale down necessarie per raggiungere le dimensioni del cluster appropriate.

Per i cluster che richiedono una maggiore stabilità, imposta un valore scaleDownFactor più basso per una velocità di scale down più lenta.

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

Configurazione di scaleUpMinWorkerFraction e scaleDownMinWorkerFraction in corso…

scaleUpMinWorkerFraction e scaleDownMinWorkerFraction vengono utilizzati con scaleUpFactor o scaleDownFactor e hanno valori predefiniti di 0.0. Rappresentano le soglie in corrispondenza delle quali il gestore della scalabilità automatica eseguirà lo scale up o fare lo scale down del cluster: l'aumento o la diminuzione del valore frazionario minimo delle dimensioni del cluster necessario per emettere una richiesta 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 impostato su 0.1, il gestore della scalabilità automatica non invierà la richiesta di fare lo scale up del cluster. Analogamente, se scaleDownMinWorkerFraction è 0.05, il gestore della scalabilità automatica non emetterà una richiesta di scale down, a meno che non debbano essere rimossi almeno 5 nodi.

Il valore predefinito 0.0 indica che non esiste una soglia.

Ti consigliamo vivamente di impostare soglie scaleDownMinWorkerFraction più elevate su cluster di grandi dimensioni (> 100 nodi) per evitare operazioni di scalabilità di piccole dimensioni e non necessarie.

Scegli un periodo di attesa

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

Il valore minimo e predefinito di cooldownPeriod è di due minuti. Se in una policy viene impostato un valore cooldownPeriod più breve, le modifiche al workload influiranno più rapidamente sulle dimensioni del cluster, ma i cluster potrebbero fare lo scale up e lo scale down inutilmente. La prassi consigliata è impostare scaleUpMinWorkerFraction e scaleDownMinWorkerFraction di una policy su un valore diverso da zero quando si utilizza un cooldownPeriod più breve. In questo modo, lo scale up o lo scale down del cluster viene eseguito solo quando la variazione nell'utilizzo delle risorse è sufficiente per giustificare un aggiornamento del cluster.

Se il tuo workload è sensibile alle variazioni delle dimensioni del cluster, puoi aumentare il periodo di attesa. Ad esempio, se esegui un job di elaborazione batch, puoi impostare il periodo di attesa su 10 minuti o più. Sperimenta diversi periodi di attesa per trovare il valore più adatto al tuo workload.

Limiti del numero di worker e pesi dei gruppi

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

Ogni gruppo ha anche un parametro denominato weight che configura il bilanciamento target tra i due gruppi. Tieni presente che questo parametro è solo un suggerimento e, 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.

Utilizza la scalabilità automatica basata sui core

Per le applicazioni ad alta intensità di CPU, una best practice consiste nell'utilizzare il Dominant Resource Calculator per l'allocazione delle risorse. Questa è la configurazione YARN predefinita a partire dalla versione immagine 2.2 di Managed Service for Apache Spark. Con le versioni immagine precedenti, Managed Service for Apache Spark configura YARN in modo che utilizzi le metriche della memoria per l'allocazione delle risorse, a meno che non imposti la seguente proprietà quando crei un cluster per configurare YARN in modo che utilizzi il Dominant Resource Calculator:

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

Metriche e log di scalabilità automatica

Le seguenti risorse e strumenti 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:

  • visualizzare le metriche utilizzate dalla scalabilità automatica
  • visualizzare il numero di Node Manager nel tuo cluster
  • capire perché la scalabilità automatica ha eseguito o meno la scalabilità del cluster autoscaling-stackdriver1 autoscaling-stackdriver2 autoscaling-stackdriver3

Cloud Logging

Utilizza Cloud Logging per visualizzare i log del gestore della scalabilità automatica di Managed Service for Apache Spark.

1) Trova i log per il tuo cluster.

autoscaling-logs-for-cluster

2) Seleziona dataproc.googleapis.com/autoscaler.

autoscaling-log-file

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

autoscaling-three-logs autoscaling-update-operation

4) Espandi il messaggio di log per visualizzare i suggerimenti di scalabilità, le metriche utilizzate per le decisioni di scalabilità, le dimensioni originali del cluster e le nuove dimensioni target del cluster.

autoscaling-recommendation-message

Contesto: scalabilità automatica con Apache Hadoop e Apache Spark

Le sezioni seguenti illustrano come la scalabilità automatica interagisce (o meno) 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 occupata dai container in esecuzione nell'intero cluster. Se sono presenti 6 container in esecuzione che possono utilizzare fino a 1 unità di risorsa, le risorse allocate sono 6.

  2. Available resource è la risorsa YARN nel cluster non utilizzata dai container allocati. Se sono presenti 10 unità di risorse in tutti i Node Manager e 6 di queste sono allocate, sono disponibili 4 risorse. Se nel cluster sono presenti risorse disponibili (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 stanno aspettando che si liberi spazio per l'esecuzione in YARN. La risorsa in attesa è diversa da zero solo se la risorsa disponibile è pari a zero o troppo piccola per essere allocata al container successivo. Se sono presenti container in attesa, la scalabilità automatica potrebbe aggiungere worker al cluster.

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

Scalabilità automatica e Hadoop MapReduce

MapReduce esegue ogni attività di mappatura e riduzione come container YARN separato. Quando un job inizia, MapReduce invia richieste di container per ogni attività di mappatura, il che comporta un picco elevato di memoria YARN in attesa. Man mano che le attività di mappatura vengono completate, la memoria in attesa diminuisce.

Quando i mapreduce.job.reduce.slowstart.completedmaps sono stati completati (95% per impostazione predefinita su Managed Service for Apache Spark), MapReduce mette in coda le richieste di container per tutti i riduttori, il che comporta 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 ci sia un lavoro in attesa sufficiente per utilizzare il nuovo worker per diversi minuti. Un buon punto di partenza è impostare scaleUpFactor su 0,05 (5%) o 0,1 (10%) della memoria in attesa.

Scalabilità automatica e Spark

Spark aggiunge un ulteriore livello di pianificazione oltre a YARN. In particolare, l'allocazione dinamica di Spark Core invia richieste a YARN per i container per eseguire gli esecutori Spark, quindi pianifica le attività Spark sui thread di questi esecutori. I cluster Managed Service for Apache Spark abilitano l'allocazione dinamica per impostazione predefinita, quindi gli esecutori vengono aggiunti e rimossi in base alle esigenze.

Spark richiede sempre i container a YARN, ma senza l'allocazione dinamica, li richiede solo all'inizio del job. Con l'allocazione dinamica, rimuoverà i container o ne richiederà di nuovi, a seconda delle necessità.

Spark inizia con un numero ridotto di esecutori (2 sui cluster con scalabilità automatica) e continua a raddoppiare il numero di esecutori finché sono presenti attività in arretrato. In questo modo, la memoria in attesa viene distribuita in modo più uniforme (meno picchi di memoria in attesa). Per i job Spark, ti consigliamo di impostare il fattore di scalabilità automatica scaleUpFactor su un numero elevato, ad esempio 1,0 (100%).

Disattivazione dell'allocazione dinamica di Spark

Se esegui job Spark separati che non usufruiscono dell'allocazione dinamica di Spark, puoi disattivare l'allocazione dinamica di Spark impostando spark.dynamicAllocation.enabled=false e spark.executor.instances. Puoi comunque utilizzare la scalabilità automatica per fare lo scale up e lo scale down dei cluster mentre vengono eseguiti i job Spark separati.

Job Spark con dati memorizzati nella cache

Imposta spark.dynamicAllocation.cachedExecutorIdleTimeout o rimuovi i set di dati dalla cache quando non sono più necessari. Per impostazione predefinita, Spark non rimuove gli esecutori che hanno 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 di 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 la dismissione controllata (scalabilità automatica gracefulDecommissionTimeout) con i job di Spark Streaming. Per rimuovere in sicurezza i worker con la scalabilità automatica, configura il checkpoint per la tolleranza di errore.

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

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

Scalabilità automatica e Spark Structured Streaming

La scalabilità automatica non è compatibile con Spark Structured Streaming, poiché Spark Structured Streaming non supporta l'allocazione dinamica (consulta SPARK-24815: Structured Streaming should support dynamic allocation).

Controlla la scalabilità automatica tramite partizionamento e parallelismo

Mentre il parallelismo viene solitamente impostato o determinato dalle risorse del cluster (ad esempio, il numero di blocchi HDFS controlla il 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 le linee guida per aiutarti a impostare il parallelismo dei job:

  • Sebbene Managed Service for Apache Spark 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 per impostazione predefinita è 200.
  • Le funzioni RDD di Spark utilizzano per impostazione predefinita spark.default.parallelism, che viene impostato sul numero di core sui nodi worker all'avvio del job. Tuttavia, tutte le funzioni RDD che creano shuffle accettano un parametro per il numero di partizioni, che sostituisce spark.default.parallelism.

Assicurati che i dati siano partizionati in modo uniforme. Se l'asimmetria delle chiavi è significativa, una o più attività potrebbero richiedere molto più tempo rispetto ad altre, con un conseguente utilizzo ridotto.

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

I cluster con scalabilità automatica hanno valori predefiniti delle proprietà del cluster che aiutano a evitare errori dei job quando i worker primari vengono rimossi o i worker secondari vengono prerilasciati. Puoi eseguire l'override di questi valori predefiniti quando crei un cluster con scalabilità automatica (consulta Proprietà del cluster).

Valori predefiniti per aumentare il numero massimo di nuovi tentativi per attività, master dell'applicazione e fasi: 

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

Valori predefiniti per reimpostare i contatori dei nuovi tentativi (utile per i job Spark Streaming a lunga esecuzione): 

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

Imposta come predefinita l'allocazione dinamica con avvio lento di Spark per iniziare con dimensioni ridotte:

spark:spark.executor.instances=2

Domande frequenti

Questa sezione contiene domande e risposte comuni sulla scalabilità automatica.

È possibile attivare la scalabilità automatica sui cluster ad alta affidabilità e sui cluster a nodo singolo?

La scalabilità automatica può essere attivata sui cluster ad alta affidabilità, ma non sui 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 durante l'ottimizzazione di una policy di scalabilità automatica. Tuttavia, queste modifiche avranno solo un effetto temporaneo e, alla fine, la scalabilità automatica ridurrà il cluster.

Invece di ridimensionare manualmente un cluster con scalabilità automatica, valuta la possibilità di:

Aggiornamento della policy di scalabilità automatica. Qualsiasi modifica apportata alla policy di scalabilità automatica influisce su tutti i cluster che utilizzano la policy (consulta Utilizzo della policy multi-cluster).

Scollegando la policy e scalando manualmente il cluster alle dimensioni preferite.

Assistenza per Managed Service for Apache Spark.

In che modo Managed Service for Apache Spark si differenzia dalla scalabilità automatica di Dataflow?

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

Il team di sviluppo di Managed Service for Apache Spark può reimpostare lo stato del cluster da ERROR a RUNNING?

In genere, no. Questa operazione richiede un intervento manuale per verificare se è sicuro reimpostare lo stato del cluster e spesso un cluster non può essere reimpostato senza altri passaggi manuali, come il riavvio del NameNode HDFS.

Managed Service for Apache Spark imposta lo stato di un cluster su ERROR quando non riesce a determinare lo stato di un cluster dopo un'operazione non riuscita. I cluster in ERROR non vengono scalati automaticamente. Di seguito sono elencate alcune delle possibili cause:

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

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

  • Errori dell'API Dataproc Control, ad esempio "Task lease expired".

Elimina e ricrea i cluster il cui stato è ERROR.

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

La seguente immagine illustra quando la scalabilità automatica annulla un'operazione di scale down (consulta anche Come funziona la scalabilità automatica).

dataproc-autoscaling-cancellation-example

Note:

  • La scalabilità automatica del cluster è abilitata solo in base alle metriche della memoria YARN.
  • T1-T9 rappresentano gli intervalli di attesa in cui 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 dismissione e dismessi.
  • Il numero di worker consigliato dal gestore della scalabilità automatica (linea nera) si basa sulle metriche di memoria YARN, sul conteggio dei worker attivi YARN e sulle impostazioni della policy di scalabilità automatica (consulta Come funziona la scalabilità automatica).
  • L'area con sfondo rosso indica il periodo in cui è in esecuzione l'operazione di scale down.
  • L'area con sfondo giallo indica il periodo in cui l'operazione di scale down è in fase di annullamento.
  • L'area con sfondo verde indica il periodo dell'operazione di scale up.

Le seguenti operazioni vengono eseguite nei seguenti orari:

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

  • T2: il gestore della scalabilità automatica continua a monitorare le metriche del cluster. Non modifica il suggerimento di scale down e l'operazione di scale down continua. Alcuni worker sono stati ritirati e altri sono in fase di ritiro (Managed Service for Apache Spark eliminerà i worker ritirati).

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

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

  • T5: la memoria in attesa di YARN aumenta e la variazione del numero di worker consigliato dal gestore della scalabilità automatica aumenta. Tuttavia, poiché il numero di worker attivi più la variazione consigliata del numero di worker è inferiore al numero di worker attivi più quelli in fase di ritiro, i criteri di annullamento non vengono soddisfatti 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 dal gestore della scalabilità automatica è ora maggiore del numero di worker attivi più quelli in fase di ritiro. I criteri di annullamento sono 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 scale down. Il gestore della scalabilità automatica non valuta né consiglia una modifica del numero di worker durante questo intervallo.

  • T8: l'annullamento dell'operazione di scale down viene completato. I worker in fase di dismissione vengono aggiunti al cluster e diventano attivi. Il gestore della scalabilità automatica rileva il completamento dell'annullamento dell'operazione di scale down e attende il periodo di valutazione successivo (T9) per calcolare il numero consigliato di worker.

  • T9: non sono presenti operazioni attive al momento T9. In base alla policy del gestore della scalabilità automatica e alle metriche YARN, il gestore della scalabilità automatica consiglia un'operazione di scale up.