Ottimizzare la scalabilità automatica orizzontale per le pipeline di streaming

Nelle pipeline di streaming con un volume elevato di dati di input, in genere esiste un compromesso tra costo e latenza. Per mantenere una bassa latenza, Dataflow deve aggiungere worker man mano che aumenta il volume di traffico. Un altro fattore è la velocità con cui la pipeline deve aumentare o diminuire in risposta alle variazioni della frequenza di dati di input.

Il ridimensionamento automatico di Dataflow ha impostazioni predefinite adatte per molti carichi di lavoro. Tuttavia, ti consigliamo di ottimizzare questo comportamento per il tuo scenario specifico. Ad esempio, potrebbe essere accettabile una latenza media più elevata per ridurre i costi oppure potresti voler scalare Dataflow più rapidamente in risposta ai picchi di traffico.

Per ottimizzare la scalabilità automatica orizzontale, puoi modificare i seguenti parametri:

Impostare l'intervallo di scalabilità automatica

Quando crei un nuovo job di streaming, puoi impostare il numero iniziale di worker e il numero massimo di worker. Per farlo, specifica le seguenti opzioni di pipeline:

Java

  • --numWorkers: il numero iniziale di worker disponibili quando la pipeline inizia a essere eseguita
  • --maxNumWorkers: il numero massimo di worker disponibili per la pipeline

Python

  • --num_workers: il numero iniziale di worker disponibili quando la pipeline inizia a essere eseguita
  • --max_num_workers: il numero massimo di worker disponibili per la pipeline

Vai

  • --num_workers: il numero iniziale di worker disponibili quando la pipeline inizia a essere eseguita
  • --max_num_workers: il numero massimo di worker disponibili per la pipeline

Per i job in streaming che utilizzano Streaming Engine, il flag --maxNumWorkers è facoltativo. Il valore predefinito è 100. Per i job di streaming che non utilizzano Streaming Engine,--maxNumWorkers è obbligatorio quando la scalabilità automatica orizzontale è abilitata.

Il valore iniziale --maxNumWorkers determina anche il numero di dischi permanenti allocati per il job. Le pipeline vengono implementate con un pool fisso di dischi permanenti, in numero uguale a --maxNumWorkers. Durante lo streaming, i dischi permanenti vengono ridistribuiti in modo che ogni worker riceva un numero uguale di dischi collegati.

Se imposti --maxNumWorkers, assicurati che il valore fornisca dischi sufficienti per la pipeline. Tieni conto della crescita futura quando imposti il valore iniziale. Per informazioni sulle prestazioni di Persistent Disk, consulta Configurare i dischi permanenti e le VM. Dataflow fattura l'utilizzo di Persistent Disk e ha quote di Compute Engine, incluse le quote di Persistent Disk.

Per impostazione predefinita, il numero minimo di worker è 1 per i job di streaming che utilizzano Streaming Engine e (maxNumWorkers/15), arrotondato per eccesso, per i job che non utilizzano Streaming Engine.

Aggiorna l'intervallo di scalabilità automatica

Per i job che utilizzano Streaming Engine, puoi modificare il numero minimo e massimo di worker, senza interrompere o sostituire il job. Per modificare questi valori, utilizza un aggiornamento dei job in esecuzione. Aggiorna le seguenti opzioni di job:

  • --min-num-workers: il numero minimo di lavoratori.

  • --max-num-workers: il numero massimo di worker.

gcloud

Utilizza il comando gcloud dataflow jobs update-options:

gcloud dataflow jobs update-options \
  --region=REGION \
  --min-num-workers=MINIMUM_WORKERS \
  --max-num-workers=MAXIMUM_WORKERS \
  JOB_ID

Sostituisci quanto segue:

  • REGION: l'ID regione dell'endpoint a livello di regione del job
  • MINIMUM_WORKERS: il numero minimo di istanze Compute Engine
  • MAXIMUM_WORKERS: il numero massimo di istanze Compute Engine
  • JOB_ID: l'ID del job da aggiornare

Puoi anche aggiornare --min-num-workers e --max-num-workers individualmente.

REST

Utilizza il metodo projects.locations.jobs.update:

PUT https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID?updateMask=runtime_updatable_params.max_num_workers,runtime_updatable_params.min_num_workers
{
  "runtime_updatable_params": {
    "min_num_workers": MINIMUM_WORKERS,
    "max_num_workers": MAXIMUM_WORKERS
  }
}

Sostituisci quanto segue:

  • PROJECT_ID: l'ID progetto Google Cloud del job Dataflow
  • REGION: l'ID regione dell'endpoint a livello di regione del job
  • JOB_ID: l'ID del job da aggiornare
  • MINIMUM_WORKERS: il numero minimo di istanze Compute Engine
  • MAXIMUM_WORKERS: il numero massimo di istanze Compute Engine

Puoi anche aggiornare min_num_workers e max_num_workers singolarmente. Specifica i parametri da aggiornare nel parametro di query updateMask e includi i valori aggiornati nel campo runtimeUpdatableParams del corpo della richiesta. Nell'esempio seguente viene aggiornato min_num_workers:

PUT https://dataflow.googleapis.com/v1b3/projects/my_project/locations/us-central1/jobs/job1?updateMask=runtime_updatable_params.min_num_workers
{
  "runtime_updatable_params": {
    "min_num_workers": 5
  }
}

Per i job che non utilizzano Streaming Engine, puoi sostituire il job esistente con un valore aggiornato di maxNumWorkers.

Se aggiorni un job di streaming che non utilizza Streaming Engine, la scalabilità automatica orizzontale del job aggiornato è disattivata per impostazione predefinita. Per mantenere attiva la scalabilità automatica, specifica --autoscalingAlgorithm e --maxNumWorkers per il job aggiornato.

Impostare l'indicazione sull'utilizzo dei lavoratori

Dataflow utilizza l'utilizzo medio della CPU come indicatore per sapere quando applicare la scalabilità automatica orizzontale. Per impostazione predefinita, Dataflow imposta un utilizzo della CPU di destinazione pari a 0,8. Quando l'utilizzo non rientra in questo intervallo, Dataflow potrebbe aggiungere o rimuovere worker.

Per un maggiore controllo sul comportamento della scalabilità automatica, puoi impostare l'utilizzo della CPU target su un valore compreso nell'intervallo [0,1, 0,9].

  • Imposta un valore di utilizzo della CPU più basso se vuoi ottenere latenze di picco inferiori. Un valore più basso consente a Dataflow di eseguire lo scaling out in modo più aggressivo in risposta all'aumento dell'utilizzo dei worker e di eseguire lo scaling down in modo più conservativo per migliorare la stabilità. Un valore più basso offre anche più spazio di manovra quando la pipeline è in esecuzione in stato stabile, in genere con una latenza coda inferiore. La latenza finale misura i tempi di attesa più lunghi prima dell'elaborazione di un nuovo record.

  • Imposta un valore più alto se vuoi risparmiare risorse e mantenere bassi i costi durante i picchi di traffico. Un valore più alto impedisce un upscaling eccessivo, a scapito di una maggiore latenza.

Per configurare il suggerimento di utilizzo quando esegui un job non basato su modello, imposta l'worker_utilization_hint opzione di servizio. Per un job modello, aggiorna l'indicazione di utilizzo, poiché le opzioni di servizio non sono supportate.

L'esempio seguente mostra come utilizzare worker_utilization_hint:

Java

--dataflowServiceOptions=worker_utilization_hint=TARGET_UTILIZATION

Sostituisci TARGET_UTILIZATION con un valore compreso nell'intervallo [0,1, 0,9].

Python

--dataflow_service_options=worker_utilization_hint=TARGET_UTILIZATION

Sostituisci TARGET_UTILIZATION con un valore compreso nell'intervallo [0,1, 0,9].

Vai

--dataflow_service_options=worker_utilization_hint=TARGET_UTILIZATION

Sostituisci TARGET_UTILIZATION con un valore compreso nell'intervallo [0,1, 0,9].

Per le nuove pipeline, ti consigliamo di eseguire il test con carichi realistici utilizzando l'impostazione predefinita. Valuta quindi il comportamento della scalabilità automatica per la tua pipeline e apporta le modifiche necessarie.

L'indicazione di utilizzo è solo uno dei fattori utilizzati da Dataflow per decidere se scalare i worker. Altri fattori come il backlog e le chiavi disponibili possono sostituire il valore dell'indizio. Inoltre, l'indicazione non è un target rigoroso. L'agente di scalabilità automatica tenta di mantenere l'utilizzo della CPU entro l'intervallo del valore del suggerimento, ma la metrica di utilizzo aggregata potrebbe essere superiore o inferiore. Per ulteriori informazioni, consulta Euristiche di scalabilità automatica in streaming.

Aggiornare l'indicazione sull'utilizzo

Per aggiornare l'indicazione di utilizzo durante l'esecuzione di un job, esegui un aggiornamento in tempo reale come segue:

gcloud

Utilizza il comando gcloud dataflow jobs update-options:

gcloud dataflow jobs update-options \
  --region=REGION \
  --worker-utilization-hint=TARGET_UTILIZATION \
  JOB_ID

Sostituisci quanto segue:

  • REGION: l'ID regione dell'endpoint a livello di regione del job
  • JOB_ID: l'ID del job da aggiornare
  • TARGET_UTILIZATION: un valore compreso nell'intervallo [0,1, 0,9]

Per reimpostare l'indicazione sull'utilizzo sul valore predefinito, utilizza il seguente comando gcloud:

gcloud dataflow jobs update-options \
  --unset-worker-utilization-hint \
  --region=REGION \
  --project=PROJECT_ID \
  JOB_ID

REST

Utilizza il metodo projects.locations.jobs.update:

PUT https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID?updateMask=runtime_updatable_params.worker_utilization_hint
{
  "runtime_updatable_params": {
    "worker_utilization_hint": TARGET_UTILIZATION
  }
}

Sostituisci quanto segue:

  • PROJECT_ID: l'ID progetto Google Cloud del job Dataflow.
  • REGION: l'ID regione dell'endpoint a livello di regione del job.
  • JOB_ID: l'ID del job da aggiornare.
  • TARGET_UTILIZATION: un valore compreso nell'intervallo [0,1, 0,9]

Algoritmi di ottimizzazione della scalabilità automatica di Streaming

Per le pipeline di streaming, lo scopo della scalabilità automatica orizzontale è ridurre al minimo il backlog, massimizzare l'utilizzo e la velocità effettiva dei worker e reagire rapidamente ai picchi di carico.

Dataflow prende in considerazione diversi fattori durante l'autoscaling, tra cui:

  • Backlog. Il tempo di coda stimato viene calcolato in base alla velocità in bit e ai byte in coda ancora da elaborare dall'origine di input. Una pipeline è considerata in backlog quando il tempo di backlog stimato rimane superiore a 15 secondi.

  • Utilizzo CPU target. Il target predefinito per l'utilizzo medio della CPU è 0,8. Puoi sostituire questo valore.

  • Chiavi disponibili. Le chiavi sono l'unità fondamentale del parallelismo in Dataflow.

In alcuni casi, Dataflow utilizza i seguenti fattori per prendere decisioni sull'autoscaling. Se questi fattori vengono utilizzati per il tuo job, puoi visualizzare queste informazioni nella scheda delle metriche Autoscaling.

  • La limitazione basata su chiavi utilizza il numero di chiavi di elaborazione ricevute dal job per calcolare il limite per i worker dell'utente, poiché ogni chiave può essere elaborata da un solo worker alla volta.

  • Attenuazione del ridimensionamento. Se Dataflow rileva che sono state prese decisioni di scalabilità automatica instabili, rallenta la frequenza dello scale down per migliorare la stabilità.

  • L'upscaling basato sulla CPU utilizza un utilizzo elevato della CPU come criterio di upscaling.

  • Per i job di streaming che non utilizzano Streaming Engine, la scalabilità potrebbe essere limitata dal numero di dischi permanenti. Per ulteriori informazioni, consulta Impostare l'intervallo di scalabilità automatica.

Upscaling. Se una pipeline di streaming rimane in coda con un parallelismo sufficiente sui worker per diversi minuti, Dataflow esegue l'upscaling. Dataflow tenta di eliminare la coda entro circa 150 secondi dall'aumento di scala, in base al throughput corrente per worker. Se è presente un backlog, ma il worker non dispone di parallelismo sufficiente per altri worker, la pipeline non viene scalata. L'aumento del numero di worker oltre il numero di chiavi disponibili per l'elaborazione parallela non consente di elaborare più rapidamente le richieste in attesa.

Riduzione Quando il gestore della scalabilità automatica prende una decisione di riduzione, il backlog è il fattore di priorità più elevato. Lo scopo del gestore della scalabilità automatica è avere un backlog di non più di 15 secondi. Se il backlog scende al di sotto di 10 secondi e l'utilizzo medio dei worker è inferiore all'utilizzo della CPU target, Dataflow esegue lo scale down. Se il backlog è accettabile, il gestore della scalabilità automatica tenta di mantenere l'utilizzo della CPU vicino a quello target. Tuttavia, se l'utilizzo è già sufficientemente vicino al target, lo scalatore automatico potrebbe mantenere invariato il numero di worker, perché ogni passaggio di riduzione ha un costo.

Streaming Engine utilizza anche una tecnica di scalabilità automatica predittiva basata sul backlog del timer. I dati illimitati in una pipeline in streaming sono suddivisi in windows raggruppati in base ai timestamp. Alla fine di una finestra, vengono attivati i timer per ogni chiave elaborata in quella finestra. L'attivazione di un timer indica che la finestra è scaduta per una determinata chiave. Streaming Engine può misurare il backlog dei timer e prevedere quanti timer verranno attivati alla fine di una finestra. Utilizzando il backlog dei timer come indicatore, Dataflow può stimare la quantità di elaborazione che deve essere eseguita quando vengono attivati i timer futuri. In base al carico futuro stimato, Dataflow esegue la scalabilità automatica in anticipo per soddisfare la domanda prevista.

Metriche

Per trovare i limiti attuali della scalabilità automatica per un job, esegui una query sulle seguenti metriche:

  • job/max_worker_instances_limit: numero massimo di worker.
  • job/min_worker_instances_limit: numero minimo di lavoratori.

Per informazioni sull'utilizzo dei worker, esegui query sulle seguenti metriche:

  • job/aggregated_worker_utilization: l'utilizzo aggregato dei lavoratori.
  • job/worker_utilization_hint: l'indicazione dell'utilizzo attuale del lavoratore.

Per ottenere informazioni sul comportamento del gestore della scalabilità automatica, esegui una query sulla seguente metrica:

  • job.worker_utilization_hint_is_actively_used: indica se il gestore della scalabilità automatica utilizza attivamente il suggerimento per l'utilizzo dei worker. Se altri fattori superano il suggerimento quando viene campionato questa metrica, il valore è false.
  • job/horizontal_worker_scaling: descrive le decisioni prese dal gestore della scalabilità automatica. Questa metrica contiene le seguenti etichette:
    • direction: specifica se il gestore della scalabilità automatica ha eseguito lo scale up, lo scale down o non ha intrapreso alcuna azione.
    • rationale: specifica la motivazione della decisione del gestore della scalabilità automatica.

Per ulteriori informazioni, consulta Metriche di Cloud Monitoring. Queste metriche vengono visualizzate anche nei grafici di monitoraggio dell'autoscaling.

Passaggi successivi