Ottimizzare la scalabilità automatica orizzontale per le pipeline di streaming

Nelle pipeline di streaming con dati di input fluttuanti o ad alto volume, spesso è necessario bilanciare la latenza di elaborazione con i costi operativi. Sebbene Dataflow fornisca impostazioni di scalabilità automatica orizzontale predefinite adatte a molti workload, il comportamento predefinito potrebbe non corrispondere ai tuoi requisiti specifici. Ad esempio, potresti voler ridurre i costi accettando una latenza media più elevata, fare lo scale up più rapidamente per gestire i picchi di traffico o applicare limiti rigorosi all'allocazione dei worker.

L'ottimizzazione della scalabilità automatica orizzontale ti consente di personalizzare il modo in cui Dataflow scala la capacità dei worker in risposta alle mutevoli esigenze dei workload. Modificando i parametri di scalabilità automatica, puoi ottimizzare la velocità effettiva della pipeline, controllare i costi di computing e raggiungere i tuoi obiettivi di latenza specifici.

Ottimizza la scalabilità automatica orizzontale

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

Imposta l'intervallo di scalabilità automatica

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

Java

  • --numWorkers: il numero iniziale di worker disponibili all'avvio dell'esecuzione della pipeline
  • --maxNumWorkers: il numero massimo di worker disponibili per la pipeline

Python

  • --num_workers: il numero iniziale di worker disponibili all'avvio dell'esecuzione della pipeline
  • --max_num_workers: il numero massimo di worker disponibili per la tua pipeline

Vai

  • --num_workers: il numero iniziale di worker disponibili all'avvio dell'esecuzione della pipeline
  • --max_num_workers: il numero massimo di worker disponibili per la tua pipeline

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

Il valore iniziale di --maxNumWorkers determina anche il numero di Persistent Disk allocati per il job. Il deployment delle pipeline viene eseguito con un pool fisso di Persistent Disk, in numero pari a --maxNumWorkers. Durante lo streaming, i Persistent Disk vengono ridistribuiti in modo che ogni worker riceva un numero uguale di dischi collegati.

Se imposti --maxNumWorkers, assicurati che il valore fornisca un numero di dischi sufficiente per la tua pipeline. Considera la crescita futura quando imposti il valore iniziale. Per informazioni sulle prestazioni di Persistent Disk, consulta Configura Persistent Disk 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 flussi di dati 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 del job in corso. Aggiorna le seguenti opzioni del job:

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

  • --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 regionale del job
  • MINIMUM_WORKERS: il numero minimo di istanze di 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 singolarmente.

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 regionale del job
  • JOB_ID: l'ID del job da aggiornare
  • MINIMUM_WORKERS: il numero minimo di istanze di 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. L'esempio seguente aggiorna 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 flussi di dati che non utilizza Streaming Engine, la scalabilità automatica orizzontale è disabilitata per impostazione predefinita nel job aggiornato. Per mantenere attiva la scalabilità automatica, specifica --autoscalingAlgorithm e --maxNumWorkers per il job aggiornato.

Imposta il suggerimento per l'utilizzo dei worker

Dataflow utilizza l'utilizzo medio della CPU come indicatore per stabilire quando applicare la scalabilità automatica orizzontale. Per impostazione predefinita, Dataflow imposta un utilizzo della CPU target di 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 inferiore se vuoi ottenere latenze di picco inferiori. Un valore più basso consente a Dataflow di fare lo scale out in modo più aggressivo in risposta al crescente utilizzo dei worker e di fare lo scale down in modo più conservativo per migliorare la stabilità. Un valore più basso offre anche un margine maggiore quando la pipeline è in esecuzione in uno stato stabile, il che generalmente si traduce in una latenza di coda inferiore. (La latenza di coda misura i tempi di attesa più lunghi prima che venga elaborato un nuovo record.)

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

Per configurare il suggerimento di utilizzo quando esegui un job non basato su template, imposta l'opzione di servizio worker_utilization_hint. Per un job basato su template, aggiorna il suggerimento 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 test con carichi realistici utilizzando l'impostazione predefinita. Poi valuta il comportamento di scalabilità automatica applicato alla tua pipeline e apporta le modifiche necessarie.

Il suggerimento sull'utilizzo è solo uno dei fattori che Dataflow utilizza per decidere se scalare i worker. Altri fattori, come il backlog e le chiavi disponibili, possono sostituire il valore del suggerimento. Inoltre, il suggerimento non è un target rigido. Il gestore della 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 saperne di più, consulta Euristica della scalabilità automatica in modalità flusso.

Aggiorna il suggerimento di utilizzo

Per aggiornare il suggerimento di utilizzo mentre un job è in esecuzione, esegui un aggiornamento in corso 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 regionale del job
  • JOB_ID: l'ID del job da aggiornare
  • TARGET_UTILIZATION: un valore compreso nell'intervallo [0.1, 0.9]

Per reimpostare il suggerimento di 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 regionale del job.
  • JOB_ID: l'ID del job da aggiornare.
  • TARGET_UTILIZATION: un valore compreso nell'intervallo [0.1, 0.9]

Imposta il suggerimento di parallelismo dei worker

Per gestire la scalabilità automatica con operazioni lunghe che dipendono meno dalle CPU, come i workload ad alta intensità di ML, puoi impostare il suggerimento di parallelismo dei worker utilizzando i suggerimenti per le risorse Apache Beam. Questi suggerimenti attivano la scalabilità automatica in una modalità diversa ottimizzata per workload che richiedono un uso intensivo della GPU o trasformazioni con tempi di elaborazione lunghi.

L'esempio seguente mostra come collegare un suggerimento di parallelismo a una trasformazione:

Java

pcoll.apply(MyCompositeTransform.of(...)
  .setResourceHints(
      ResourceHints.create()
          .withMaxActiveBundlesPerWorker(TARGET_PARALLELISM_PER_WORKER)))

Sostituisci TARGET_PARALLELISM_PER_WORKER con un valore appropriato per il tuo caso d'uso. Per indicazioni generali, consulta la sezione su come scegliere un buon valore iniziale.

Python

pcoll | MyPTransform().with_resource_hints(
  max_active_bundles_per_worker="TARGET_PARALLELISM_PER_WORKER")

Sostituisci TARGET_PARALLELISM_PER_WORKER con un valore appropriato per il tuo caso d'uso. Per indicazioni generali, consulta la sezione su come scegliere un buon valore iniziale.

Scegli il valore del suggerimento per il parallelismo dei worker

Per i casi d'uso di ML, un buon valore iniziale è equivalente al numero di modelli in esecuzione in parallelo all'interno di ogni worker. Questo valore è limitato dalla capacità degli acceleratori sul worker e dalle dimensioni del modello.

Per altri casi d'uso, la pipeline è limitata dalla memoria o dalla CPU. Per le pipeline con vincoli di memoria, utilizza il limite di memoria per calcolare l'elaborazione parallela massima. Per le pipeline con vincoli di CPU, è consigliabile mantenere la policy di scalabilità automatica predefinita, anziché fornire un suggerimento di parallelismo.

È possibile ottimizzare il valore per soddisfare le esigenze di elaborazione di altre fasi, ad esempio la scrittura in un sink. Aumentare il valore di 1 o 2 quando il parallelismo del modello è 2 aiuta a riconoscere il tempo di elaborazione più rapido della scrittura nel sink, offrendo più margine per tenere conto dell'elaborazione eseguita in altre fasi. Se la pipeline non prevede lo shuffling e le trasformazioni sono fuse insieme in un'unica fase, non è necessario modificare il valore per altre trasformazioni.

Questo valore può anche essere modificato per simulare gli effetti di ritardi accettabili nel backlog. Ad esempio, se accetti un ritardo massimo di 10 minuti e il tempo di elaborazione medio del tuo modello è di 1 minuto, puoi scegliere di aumentare il valore di 1, supponendo che il numero massimo di worker sia impostato su 10.

Euristica di scalabilità automatica ad alta intensità di GPU

Nell'impostazione ad alta intensità di GPU indicata tramite l'impostazione del suggerimento di parallelismo, Dataflow tiene conto di diversi fattori durante la scalabilità automatica. Questi fattori includono:

  • Chiavi disponibili. Le chiavi sono l'unità fondamentale di parallelismo in Dataflow.
  • Numero massimo di bundle attivi per worker. Questo suggerisce il numero massimo ideale di parallelismo di elaborazione all'interno del worker.

L'idea generale alla base delle decisioni di scalabilità è calcolare i worker necessari per gestire il carico attuale, come indicato dalle chiavi disponibili. Ad esempio, se sono disponibili 100 chiavi da elaborare e il parallelismo massimo per worker è 10, dovresti puntare ad avere 10 worker in totale.

Se la pipeline è complessa e presenta sia un workload pesante che richiede un uso intensivo della GPU sia numerose trasformazioni che richiedono un uso intensivo della CPU, ti consigliamo di attivare l'adattamento. In questo modo, il servizio può distinguere chiaramente tra il lavoro che richiede un uso intensivo della CPU e quello che richiede un uso intensivo della GPU, quindi scalare di conseguenza ogni pool di worker.

Imposta il livello di latenza

La scalabilità automatica orizzontale di Dataflow utilizza la durata del backlog per determinare quando scalare i worker. Per impostazione predefinita, Dataflow punta a una durata del backlog inferiore a 15 secondi, adatta per i workload in tempo reale, ma che può comportare costi delle risorse più elevati.

Per i job di flussi di dati che utilizzano Streaming Engine, puoi configurare la soglia di backlog scegliendo un livello di latenza. Un livello di latenza viene mappato a una durata del backlog target, consentendoti di configurare la sensibilità del gestore della scalabilità automatica alla durata del backlog. Una decisione di scalabilità automatica viene attivata solo quando la durata del backlog supera il livello selezionato per diversi minuti.

La tabella seguente descrive i livelli di latenza disponibili:

Livello di latenza Tempo di backlog stimato consentito Caso d'uso previsto
low_latency Meno di 15 secondi Dà la priorità allo scale up rapido per ridurre al minimo il backlog. Valore predefinito selezionato se non viene specificato alcun livello
medium_latency Meno di 2 minuti Una via di mezzo equilibrata tra latenza e costi delle risorse.
high_latency Meno di 10 minuti Dà la priorità ai risparmi sui costi consentendo l'accumulo di backlog. Progettati per workload con tolleranza di ritardo.

Per configurare il livello di latenza quando crei un job, imposta l'opzione della pipeline enable_latency_tier_lightweight_update:

--experiments=enable_latency_tier_lightweight_update=TIER

Sostituisci TIER con uno dei livelli di latenza disponibili (ad esempio medium_latency).

Aggiorna il livello di latenza

Per aggiornare il livello di latenza mentre un job è in esecuzione, esegui un aggiornamento in corso utilizzando il comando gcloud dataflow jobs update-options:

gcloud dataflow jobs update-options \
  --region=REGION \
  --latency-tier=TIER \
  JOB_ID

Sostituisci quanto segue:

  • REGION: l'ID regione dell'endpoint regionale del job
  • JOB_ID: l'ID del job da aggiornare
  • TIER: il livello di latenza target. Uno dei seguenti valori: low_latency, medium_latency o high_latency

Per reimpostare il livello di latenza e ripristinare il comportamento predefinito a bassa latenza, utilizza il seguente comando:

gcloud dataflow jobs update-options \
  --unset-latency-tier \
  --region=REGION \
  JOB_ID

Pianifica le modifiche alla scalabilità automatica

Per i job di flussi di dati che utilizzano Streaming Engine, puoi configurare pianificazioni di scalabilità automatica per applicare automaticamente gli override dei parametri durante finestre temporali ricorrenti. Ciò è utile per anticipare variazioni prevedibili del carico di lavoro, ad esempio fare lo scale up durante l'orario di apertura o fare lo scale down durante i periodi di inattività, senza aggiornare manualmente il job ogni volta. Per ogni job è possibile specificare fino a 10 pianificazioni.

Una pianificazione di scalabilità automatica specifica:

  • Inizio pianificazione: un'espressione crontab che determina quando viene attivata la pianificazione.
  • Durata: il periodo di tempo in secondi per cui gli override rimangono attivi. Deve essere di almeno 5 minuti.
  • Fuso orario: un fuso orario facoltativo del database tz (ad esempio America/New_York). Il valore predefinito è UTC.
  • Priority: una priorità facoltativa per la pianificazione, dove i valori più alti indicano una priorità più alta. Le pianificazioni con priorità più alta vengono applicate per prime. Se una pianificazione è già attiva quando ne viene applicata una con priorità più alta, quest'ultima sostituisce la pianificazione esistente. Se due pianificazioni hanno la stessa priorità, viene applicata quella modificata più di recente.
  • Override dei parametri: le impostazioni di scalabilità automatica da utilizzare durante la finestra attiva, tra cui il numero minimo di worker, il numero massimo di worker e il suggerimento per l'utilizzo dei worker.

Al termine della finestra di programmazione, Dataflow ripristina automaticamente il comportamento di scalabilità automatica alle impostazioni predefinite del job. Puoi definire più pianificazioni per un singolo job.

Imposta una pianificazione di scalabilità automatica all'avvio del job

Per impostare una pianificazione di scalabilità automatica iniziale quando avvii un nuovo job di flussi di dati, specifica l'opzione della pipeline dell'esperimento autoscaling_schedule o includi il campo runtimeUpdatableParams.schedules quando crei un job utilizzando l'API REST.

Specifica l'esperimento autoscaling_schedule come stringa separata da punto e virgola di coppie key:value con i seguenti attributi:

Obbligatorio:

  • name: un identificatore univoco per la pianificazione
  • crontab: un'espressione crontab che specifica quando viene attivata la pianificazione
  • duration: la durata in secondi per cui la pianificazione rimane attiva

(Facoltativo)

  • timezone: il fuso orario del database tz (ad esempio America/New_York). Il valore predefinito è UTC
  • min_workers: il numero minimo di worker da allocare durante la pianificazione
  • max_workers: il numero massimo di worker da allocare durante la pianificazione
  • cpu_utilization_target: l'utilizzo della CPU dei worker target durante la pianificazione, nell'intervallo [0,1, 0,9]
  • latency_tier: il livello di latenza target (low_latency, medium_latency o high_latency)

  • priority: la priorità della pianificazione, dove i valori più alti indicano una priorità più alta

Java

--experiments=autoscaling_schedule="name=SCHEDULE_NAME;crontab=CRONTAB_EXPRESSION;duration=DURATION_SECONDS;timezone=TIMEZONE;min_workers=MINIMUM_WORKERS;max_workers=MAXIMUM_WORKERS;cpu_utilization_target=TARGET_UTILIZATION;latency_tier=LATENCY_TIER;priority=PRIORITY"

Python

--experiments=autoscaling_schedule="name=SCHEDULE_NAME;crontab=CRONTAB_EXPRESSION;duration=DURATION_SECONDS;timezone=TIMEZONE;min_workers=MINIMUM_WORKERS;max_workers=MAXIMUM_WORKERS;cpu_utilization_target=TARGET_UTILIZATION;latency_tier=LATENCY_TIER;priority=PRIORITY"

Vai

--experiments=autoscaling_schedule="name=SCHEDULE_NAME;crontab=CRONTAB_EXPRESSION;duration=DURATION_SECONDS;timezone=TIMEZONE;min_workers=MINIMUM_WORKERS;max_workers=MAXIMUM_WORKERS;cpu_utilization_target=TARGET_UTILIZATION;latency_tier=LATENCY_TIER;priority=PRIORITY"

REST

Utilizza il metodo projects.locations.jobs.create e includi l'oggetto runtimeUpdatableParams.schedules nel corpo della richiesta:

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs
{
"name": "JOB_NAME",
"environment": { ... },
"runtimeUpdatableParams": {
  "schedules": [
    {
      "name": "SCHEDULE_NAME",
      "crontab": "CRONTAB_EXPRESSION",
      "duration": "DURATION_SECONDSs",
      "timeZone": "TIMEZONE",
      "priority": PRIORITY,
      "parameters": {
        "minWorkerCount": MINIMUM_WORKERS,
        "maxWorkerCount": MAXIMUM_WORKERS,
        "cpuUtilizationTarget": TARGET_UTILIZATION,
        "latencyTarget": LATENCY_TIER
      }
    }
  ]
}
}

Crea o aggiorna una pianificazione di scalabilità automatica per un job in esecuzione

Per creare o aggiornare una pianificazione di scalabilità automatica per un job in esecuzione, esegui un aggiornamento in corso:

gcloud

Utilizza il comando gcloud dataflow jobs update-options con il flag --set-schedule:

gcloud dataflow jobs update-options \
  --region=REGION \
  --set-schedule=SCHEDULE_NAME \
  --autoscaling-schedule="CRONTAB_EXPRESSION" \
  --autoscaling-schedule-duration-seconds=DURATION_SECONDS \
  --autoscaling-schedule-timezone=TIMEZONE \
  --autoscaling-min-workers=MINIMUM_WORKERS \
  --autoscaling-max-workers=MAXIMUM_WORKERS \
  --autoscaling-worker-utilization-hint=TARGET_UTILIZATION \
  --autoscaling-latency-tier=LATENCY_TIER \
  --autoscaling-schedule-priority=PRIORITY \
  JOB_ID

Sostituisci quanto segue:

  • REGION: l'ID regione dell'endpoint regionale del job
  • SCHEDULE_NAME: un identificatore univoco per la pianificazione
  • CRONTAB_EXPRESSION: un'espressione crontab che specifica quando viene attivata la pianificazione (ad esempio, "0 0 * * FRI")
  • DURATION_SECONDS: la durata in secondi per cui la pianificazione è attiva (ad esempio 3600)
  • TIMEZONE: il fuso orario per la pianificazione dal database tz (ad esempio, America/New_York). Se omesso, il valore predefinito è UTC.
  • MINIMUM_WORKERS: il numero minimo di worker da allocare durante la pianificazione
  • MAXIMUM_WORKERS: il numero massimo di worker da allocare durante la pianificazione
  • LATENCY_TIER: il livello di latenza target (low_latency, medium_latency o high_latency)
  • JOB_ID: l'ID del job da aggiornare

Quando crei una nuova pianificazione, sono obbligatori sia --autoscaling-schedule sia --autoscaling-schedule-duration-seconds. Per modificare una pianificazione esistente, specifica --set-schedule e solo i parametri che vuoi aggiornare.

REST

Utilizza il metodo projects.locations.jobs.update:

PUT https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID?updateMask=runtimeUpdatableParams.schedules
{
  "runtimeUpdatableParams": {
    "schedules": [
      {
        "name": "SCHEDULE_NAME",
        "crontab": "CRONTAB_EXPRESSION",
        "duration": "DURATION_SECONDSs",
        "timeZone": "TIMEZONE",
        "priority": PRIORITY,
        "parameters": {
          "minWorkerCount": MINIMUM_WORKERS,
          "maxWorkerCount": MAXIMUM_WORKERS,
          "cpuUtilizationTarget": TARGET_UTILIZATION,
          "latencyTarget": LATENCY_TIER
        }
      }
    ]
  }
}

Sostituisci quanto segue:

  • PROJECT_ID: l'ID progetto Google Cloud del job Dataflow
  • REGION: l'ID regione dell'endpoint regionale del job
  • JOB_ID: l'ID del job da aggiornare
  • SCHEDULE_NAME: un identificatore univoco per la pianificazione
  • CRONTAB_EXPRESSION: un'espressione crontab che specifica quando viene attivata la pianificazione
  • DURATION_SECONDS: la durata in secondi per cui la pianificazione è attiva (ad esempio 3600)
  • TIMEZONE: il fuso orario per la pianificazione dal database tz (ad esempio, America/New_York)
  • MINIMUM_WORKERS: il numero minimo di worker da allocare durante la pianificazione
  • MAXIMUM_WORKERS: il numero massimo di worker da allocare durante la pianificazione
  • TARGET_UTILIZATION: un valore compreso nell'intervallo [0.1, 0.9]

Rimuovi una pianificazione della scalabilità automatica

Per rimuovere una pianificazione di scalabilità automatica specifica o tutte le pianificazioni da un job, esegui un aggiornamento in corso:

gcloud

Per rimuovere una singola pianificazione, specifica il flag --unset-schedule:

gcloud dataflow jobs update-options \
  --region=REGION \
  --unset-schedule=SCHEDULE_NAME \
  JOB_ID

Per rimuovere tutte le pianificazioni da un job, specifica il flag --unset-schedules:

gcloud dataflow jobs update-options \
  --region=REGION \
  --unset-schedules \
  JOB_ID

REST

Utilizza il metodo projects.locations.jobs.update con updateMask=runtimeUpdatableParams.schedules, omettendo la pianificazione dall'array schedules o fornendo un elenco schedules vuoto per rimuovere tutte le pianificazioni:

PUT https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID?updateMask=runtimeUpdatableParams.schedules
{
  "runtimeUpdatableParams": {
    "schedules": []
  }
}

Per risolvere i problemi relativi alle pianificazioni di scalabilità automatica, consulta Risolvi i problemi relativi alle pianificazioni di scalabilità automatica.

Euristica della scalabilità automatica in modalità flusso

Per le pipeline di streaming, l'obiettivo della scalabilità automatica orizzontale è ridurre al minimo il backlog, massimizzando al contempo l'utilizzo e il throughput dei worker e reagendo rapidamente ai picchi di carico.

Dataflow prende in considerazione diversi fattori durante la scalabilità automatica, tra cui:

  • Backlog. Il tempo di backlog stimato viene calcolato in base al throughput e ai byte di backlog ancora da elaborare dall'origine di input. Una pipeline è considerata in arretrato quando il tempo di arretrato stimato rimane al di sopra della soglia configurata. Per impostazione predefinita, questa soglia è di 15 secondi. Per saperne di più, consulta Imposta il livello di latenza.

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

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

In alcuni casi, Dataflow utilizza i seguenti fattori per le decisioni di scalabilità automatica. Se questi fattori vengono utilizzati per il tuo job, puoi visualizzare queste informazioni nella scheda delle metriche di scalabilità automatica.

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

  • Smorzamento dello scale down. Se Dataflow rileva che sono state prese decisioni di scalabilità automatica instabili, rallenta la velocità di 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 flussi di dati che non utilizzano Streaming Engine, la scalabilità potrebbe essere limitata dal numero di Persistent Disk. Per saperne di più, consulta Imposta l'intervallo di scalabilità automatica.

  • Scalabilità automatica ad alta intensità di GPU, se abilitata impostando il suggerimento di parallelismo dei worker. Per saperne di più, consulta Euristica di scalabilità automatica ad alta intensità di GPU

Upscaling. Se una pipeline di streaming rimane in arretrato con un parallelismo sufficiente sui worker per diversi minuti, Dataflow esegue lo scale up. Dataflow tenta di cancellare il backlog entro circa 150 secondi dallo scale up, in base al throughput attuale per worker. Se è presente un backlog, ma il worker non ha un parallelismo sufficiente per worker aggiuntivi, lo scale up della pipeline non viene eseguito. (La scalabilità del numero di worker oltre il numero di chiavi disponibili per l'elaborazione parallela non aiuta a elaborare più velocemente l'arretrato.)

Scale down Quando il gestore della scalabilità automatica prende una decisione di scale down, il backlog è il fattore con la priorità più alta. Il gestore della scalabilità automatica punta a un backlog basato sul livello di latenza. Se il backlog scende sufficientemente al di sotto del target per diversi minuti e l'utilizzo medio dei worker è inferiore al target di utilizzo della CPU, Dataflow esegue lo scale down. Se il backlog è accettabile, il gestore della scalabilità automatica tenta di mantenere l'utilizzo della CPU vicino all'utilizzo della CPU target. Tuttavia, se l'utilizzo è già sufficientemente vicino al target, il gestore della scalabilità automatica potrebbe mantenere invariato il numero di worker, perché ogni passaggio di scale down ha un costo.

Streaming Engine utilizza anche una tecnica di scalabilità automatica predittiva basata sul backlog del timer. I dati illimitati in una pipeline di streaming sono divisi in finestre raggruppate per timestamp. Al termine di una finestra, i timer vengono attivati 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 avvenire quando i timer futuri vengono attivati. In base al carico futuro stimato, Dataflow esegue la scalabilità automatica in anticipo per soddisfare la domanda prevista.

Metriche

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

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

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

  • job/aggregated_worker_utilization: l'utilizzo aggregato dei worker.
  • job/worker_utilization_hint: il suggerimento sull'utilizzo attuale dei worker.

Per ottenere insight 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 eseguono l'override del suggerimento quando viene eseguito il campionamento di 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 nessuna azione.
    • rationale: specifica la logica di fondo della decisione del gestore della scalabilità automatica.

Per saperne di più, consulta Metriche di Cloud Monitoring. Queste metriche vengono visualizzate anche nei grafici di monitoraggio della scalabilità automatica.

Passaggi successivi