Aggiornare una pipeline esistente

Questo documento descrive come aggiornare un job di streaming in corso. Potresti voler aggiornare il job Dataflow esistente per i seguenti motivi:

• Vuoi migliorare o ottimizzare il codice della pipeline.
• Vuoi correggere i bug nel codice della pipeline.
• Vuoi aggiornare la pipeline per gestire le modifiche al formato dei dati o per tenere conto della versione o di altre modifiche nell'origine dati.
• Vuoi applicare una patch a una vulnerabilità di sicurezza relativa al sistema operativo ottimizzato per i container per tutti i worker Dataflow.
• Vuoi scalare una pipeline Apache Beam in streaming per utilizzare un numero diverso di worker.

Puoi aggiornare i job in due modi:

• Aggiornamento dei job in esecuzione: per i job di streaming che utilizzano Streaming Engine, puoi aggiornare le opzioni dei job min-num-workers e max-num-workers senza interrompere il job o modificare l'ID job.
• Job sostitutivo: per eseguire il codice aggiornato della pipeline o per aggiornare le opzioni di job non supportate dagli aggiornamenti in corso, avvia un nuovo job che sostituisca quello esistente. Per verificare se un job sostitutivo è valido, prima di lanciare il nuovo job, convalida il relativo grafo del job.

Quando aggiorni il job, il servizio Dataflow esegue un controllo di compatibilità tra il job in esecuzione e il potenziale job sostitutivo. Il controllo di compatibilità garantisce che elementi come informazioni sullo stato intermedio e dati in buffer possano essere trasferiti dal job precedente al job sostitutivo.

Puoi anche utilizzare l'infrastruttura di logging integrata dell'SDK Apache Beam per registrare le informazioni quando aggiorni il job. Per ulteriori informazioni, consulta Utilizzare i log della pipeline. Per identificare i problemi con il codice della pipeline, utilizza il livello di logging DEBUG.

• Per istruzioni su come aggiornare i job di streaming che utilizzano i modelli classici, consulta Aggiornare un job di streaming con modello personalizzato.
• Per istruzioni su come aggiornare i job di streaming che utilizzano i modelli flessibili, segui le istruzioni dell'interfaccia a riga della gcloud CLI riportate in questa pagina o consulta Aggiornare un job con modello flessibile.

Aggiornamento dell'opzione di job in esecuzione

Per un job di inserimento flussi che utilizza Streaming Engine, puoi aggiornare le seguenti opzioni di job senza interrompere il job o modificare l'ID job:

• min-num-workers: il numero minimo di istanze Compute Engine.
• max-num-workers: il numero massimo di istanze Compute Engine.
• worker-utilization-hint: il utilizzo della CPU target, nell'intervallo [0,1, 0,9]

Per altri aggiornamenti dei job, devi sostituire il job corrente con quello aggiornato. Per ulteriori informazioni, consulta Lanciare un job sostitutivo.

Eseguire un aggiornamento in corso

Per eseguire un aggiornamento delle opzioni di job in esecuzione, svolgi i seguenti passaggi.

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

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

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
  }
}

Un job deve essere in stato di esecuzione per essere idoneo per gli aggiornamenti in corso. Viene visualizzato un errore se il job non è stato avviato o è già stato annullato. Analogamente, se avvii un job sostitutivo, attendi che inizi a essere eseguito prima di inviare aggiornamenti in corso al nuovo job.

Dopo aver inviato una richiesta di aggiornamento, ti consigliamo di attendere il completamento della richiesta prima di inviare un altro aggiornamento. Visualizza i log del job per vedere quando la richiesta viene completata.

Convalida un job di sostituzione

Per verificare se un job sostitutivo è valido, prima di lanciare il nuovo job, convalida il relativo grafo del job. In Dataflow, un grafo di job è una rappresentazione grafica di una pipeline. Con la convalida del grafico dei job, riduci il rischio che la pipeline riscontri errori o arresti anomali dopo l'aggiornamento. Inoltre, puoi convalidare gli aggiornamenti senza dover arrestare il job originale, in modo che non si verifichino tempi di inattività.

Per convalidare il grafo dei job, segui i passaggi per avviare un job sostitutivo. Includi l'opzione del servizio Dataflow graph_validate_only nel comando di aggiornamento.

gcloud dataflow flex-template run JOB_NAME --additional-experiments=graph_validate_only

{
  additionalExperiments : ["graph_validate_only"]
  ...
}

L'opzione di servizio graph_validate_only convalida solo gli aggiornamenti della pipeline. Non utilizzare questa opzione per creare o avviare le pipeline. Per aggiornare la pipeline, avvia un job sostitutivo senza l'opzione di servizio graph_validate_only.

Quando la convalida del grafico del job ha esito positivo, lo stato del job e i log del job mostrano i seguenti stati:

• Lo stato del job è JOB_STATE_DONE.
• Nella console Google Cloud, lo stato del job è Succeeded.

• Nei log dei job viene visualizzato il seguente messaggio:

  Workflow job: JOB_ID succeeded validation. Marking graph_validate_only job as Done.
  

Quando la convalida del grafico del job non va a buon fine, lo stato del job e i log del job mostrano i seguenti stati:

• Lo stato del job è JOB_STATE_FAILED.
• Nella console Google Cloud, lo stato del job è Failed.
• Nei log dei job viene visualizzato un messaggio che descrive l'errore di incompatibilità. I contenuti del messaggio dipendono dall'errore.

Avvia un job sostitutivo

Potresti sostituire un job esistente per i seguenti motivi:

• Per eseguire il codice aggiornato della pipeline.
• Per aggiornare le opzioni di job che non supportano gli aggiornamenti in tempo reale.

Per verificare se un job sostitutivo è valido, prima di lanciare il nuovo job, convalida il relativo grafo dei job.

Quando avvii un job di sostituzione, imposta le seguenti opzioni della pipeline per eseguire il processo di aggiornamento, oltre alle opzioni standard del job:

Specifica il nome del job sostitutivo

Per trovare il valore corretto del nome del job, seleziona il job precedente nell'Interfaccia di monitoraggio Dataflow. Quindi, nel riquadro laterale Informazioni job, individua il campo Nome job:

In alternativa, esegui una query su un elenco di job esistenti utilizzando l'interfaccia a riga di comando Dataflow. Inserisci il comando gcloud dataflow jobs list nella finestra della shell o del terminale per ottenere un elenco di job Dataflow nel tuo progetto Google Cloud e individua il campo NAME per il job che vuoi sostituire:

JOB_ID                                    NAME                        TYPE       CREATION_TIME        STATE    REGION
2020-12-28_12_01_09-yourdataflowjobid     ps-topic                    Streaming  2020-12-28 20:01:10  Running  us-central1

Creare una mappatura della trasformazione

Se la pipeline di sostituzione modifica i nomi delle trasformazioni rispetto a quelli della pipeline precedente, il servizio Dataflow richiede una mappatura delle trasformazioni. La mappatura delle trasformazioni mappa le trasformazioni denominate nel codice della pipeline precedente ai nomi nel codice della pipeline di sostituzione.

--transformNameMapping= . 
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

--transformNameMapping='{"oldTransform1":"newTransform1",...}'

--transform_name_mapping= .
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

--transform_name_mapping='{"oldTransform1":"newTransform1",...}'

--transform_name_mapping= .
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

--transform_name_mapping='{"oldTransform1":"newTransform1",...}'

--transform-name-mappings= .
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

--transform-name-mappings='{"oldTransform1":"newTransform1",...}'

"transformNameMapping": {
  oldTransform1: newTransform1,
  oldTransform2: newTransform2,
  ...
}

Determina i nomi delle trasformazioni

Il nome della trasformazione in ogni istanza della mappa è il nome che hai fornito quando hai applicato la trasformazione nella pipeline. Ad esempio:

  .apply("FormatResults", ParDo
    .of(new DoFn<KV<String, Long>>, String>() {
      ...
     }
  }))

  | 'FormatResults' >> beam.ParDo(MyDoFn())

  // In Go, this is always the package-qualified name of the DoFn itself.
  // For example, if the FormatResults DoFn is in the main package, its name
  // is "main.FormatResults".
  beam.ParDo(s, FormatResults, results)

Puoi anche ottenere i nomi delle trasformazioni per il job precedente esaminando il gráfo di esecuzione del job nell'interfaccia di monitoraggio di Dataflow:

Denominazione delle trasformazioni composite

I nomi delle trasformazioni sono gerarchici, in base alla gerarchia delle trasformazioni nella pipeline. Se la pipeline contiene una trasformazione composita, le trasformazioni nidificate vengono denominate in base alla trasformazione contenente. Ad esempio, supponiamo che la pipeline contenga una trasformazione composita denominata CountWidgets, che contiene una trasformazione interna denominata Parse. Il nome completo della trasformazione è CountWidgets/Parse e devi specificarlo nella mappatura della trasformazione.

Se la nuova pipeline mappa una trasformazione composita a un nome diverso, anche tutte le trasformazioni nidificate vengono rinominate automaticamente. Devi specificare i nomi modificati per le trasformazioni interne nella mappatura delle trasformazioni.

Ristruttura la gerarchia delle trasformazioni

Se la pipeline di sostituzione utilizza una gerarchia di trasformazione diversa rispetto alla precedente, devi dichiarare esplicitamente la mappatura. Potresti avere una gerarchia di trasformazioni diversa perché hai sottoposto a refactoring le trasformazioni composite o perché la tua pipeline dipende da una trasformazione composita di una libreria modificata.

Ad esempio, la pipeline precedente applicava una trasformazione composita, CountWidgets, che conteneva una trasformazione interna denominata Parse. La pipeline di sostituzione esegue il refactoring di CountWidgets e nidifica Parse all'interno di un'altra trasformazione denominata Scan. Affinché l'aggiornamento vada a buon fine, devi mappare esplicitamente il nome della trasformazione completa nella pipeline precedente (CountWidgets/Parse) al nome della trasformazione nella nuova pipeline (CountWidgets/Scan/Parse):

--transformNameMapping={"CountWidgets/Parse":"CountWidgets/Scan/Parse"}

--transformNameMapping={"CountWidgets/Parse":""}

--transform_name_mapping={"CountWidgets/Parse":"CountWidgets/Scan/Parse"}

--transform_name_mapping={"CountWidgets/Parse":""}

--transform_name_mapping={"CountWidgets/main.Parse":"CountWidgets/Scan/main.Parse"}

--transform_name_mapping={"CountWidgets/main.Parse":""}

--transform-name-mappings={"CountWidgets/Parse":"CountWidgets/Scan/Parse"}

--transform-name-mappings={"CountWidgets/main.Parse":""}

"transformNameMapping": {
  CountWidgets/Parse: CountWidgets/Scan/Parse
}

"transformNameMapping": {
  CountWidgets/main.Parse: null
}

Effetti della sostituzione di un job

Quando sostituisci un job esistente, un nuovo job esegue il codice della pipeline aggiornato. Il servizio Dataflow conserva il nome del job, ma esegue il job di sostituzione con un ID job aggiornato. Questo processo potrebbe causare un tempo di riposo mentre il job esistente si arresta, viene eseguito il controllo di compatibilità e viene avviato il nuovo job.

Il job di sostituzione conserva i seguenti elementi:

• Dati dello stato intermedio del job precedente. Le cache in memoria non vengono salvate.
• Record di dati memorizzati nella cache o metadati attualmente "in transito" del job precedente. Ad esempio, alcuni record della pipeline potrebbero essere memorizzati nella memoria intermedia in attesa della risoluzione di una finestra.
• Aggiornamenti delle opzioni di job in esecuzione applicati al job precedente.

Dati dello stato intermedio

I dati dello stato intermedio del job precedente vengono conservati. I dati di stato non includono le cache in memoria. Se vuoi conservare i dati della cache in memoria quando aggiorni la pipeline, come soluzione alternativa, esegui il refactoring della pipeline per convertire le cache in dati di stato o in input secondari. Per ulteriori informazioni sull'utilizzo degli input laterali, consulta Pattern di input laterali nella documentazione di Apache Beam.

Le pipeline di streaming hanno limiti di dimensioni per ValueState e per gli input laterali. Di conseguenza, se hai cache di grandi dimensioni che vuoi conservare, potresti dover utilizzare uno spazio di archiviazione esterno, come Memorystore o Bigtable.

Dati in volo

I dati "in transito" vengono comunque elaborati dalle trasformazioni nella nuova pipeline. Tuttavia, le trasformazioni aggiuntive aggiunte al codice della pipeline di sostituzione possono o meno essere applicate, a seconda di dove vengono memorizzati in memoria i record. In questo esempio, la pipeline esistente contiene le seguenti trasformazioni:

  p.apply("Read", ReadStrings())
   .apply("Format", FormatStrings());

  p | 'Read' >> beam.io.ReadFromPubSub(subscription=known_args.input_subscription)
    | 'Format' >> FormatStrings()

   beam.ParDo(s, ReadStrings)
   beam.ParDo(s, FormatStrings)

Puoi sostituire il job con il nuovo codice della pipeline come segue:

  p.apply("Read", ReadStrings())
   .apply("Remove", RemoveStringsStartingWithA())
   .apply("Format", FormatStrings());

  p | 'Read' >> beam.io.ReadFromPubSub(subscription=known_args.input_subscription)
    | 'Remove' >> RemoveStringsStartingWithA()
    | 'Format' >> FormatStrings()

  beam.ParDo(s, ReadStrings)
  beam.ParDo(s, RemoveStringsStartingWithA)
  beam.ParDo(s, FormatStrings)

Anche se aggiungi una trasformazione per filtrare le stringhe che iniziano con la lettera "A", la trasformazione successiva (FormatStrings) potrebbe comunque vedere stringhe in buffer o in transito che iniziano con "A" e che sono state trasferite dal job precedente.

Modificare la finestra

Puoi modificare le strategie di definizione delle finestre e di attivazione per gli elementi PCollection nella pipeline di sostituzione, ma fai attenzione. La modifica delle strategie di finestratura o attivazione non influisce sui dati già memorizzati nella cache o in fase di elaborazione.

Ti consigliamo di provare solo piccole modifiche alla finestra temporale della pipeline, come la modifica della durata delle finestre temporali fisse o scorrevoli. Apportare modifiche sostanziali alle finestre o agli attivatori, ad esempio modificare l'algoritmo di definizione delle finestre, potrebbe avere risultati imprevedibili sull'output della pipeline.

Verifica della compatibilità dei job

Quando avvii il job sostitutivo, il servizio Dataflow esegue un controllo di compatibilità tra il job sostitutivo e il job precedente. Se il controllo di compatibilità va a buon fine, il job precedente viene interrotto. Il job sostitutivo viene quindi avviato nel servizio Dataflow mantenendo lo stesso nome. Se il controllo di compatibilità non va a buon fine, il job precedente continua a essere eseguito nel servizio Dataflow e il job sostitutivo restituisce un errore.

Il controllo di compatibilità utilizza la mappatura delle trasformazioni fornita per garantire che Dataflow possa trasferire i dati dello stato intermedio dai passaggi del job precedente al job sostitutivo. Il controllo di compatibilità garantisce inoltre che gli PCollection nella pipeline utilizzino gli stessi codificatori. La modifica di un Coder può causare il fallimento del controllo di compatibilità perché eventuali dati in transito o record in buffer potrebbero non essere serializzati correttamente nella pipeline di sostituzione.

Evitare interruzioni della compatibilità

Alcune differenze tra la pipeline precedente e quella di sostituzione possono causare il fallimento del controllo di compatibilità. Queste differenze includono:

• Modifica del grafico della pipeline senza fornire una mappatura. Quando aggiorni un job, Dataflow tenta di abbinare le trasformazioni del job precedente a quelle del job sostitutivo. Questa procedura di corrispondenza consente a Dataflow di trasferire i dati dello stato intermedio per ogni passaggio. Se rinomini o rimuovi i passaggi, devi fornire una mappatura delle trasformazioni in modo che Dataflow possa abbinare i dati di stato di conseguenza.
• Modifica degli input laterali per un passaggio. L'aggiunta o la rimozione di input secondari da una trasformazione nella pipeline di sostituzione comporta il fallimento del controllo di compatibilità.
• Modificare il programmatore per un passaggio. Quando aggiorni un job, Dataflow conserva tutti i record di dati attualmente nel buffer e li gestisce nel job di sostituzione. Ad esempio, i dati nel buffer potrebbero verificarsi durante la risoluzione della definizione delle finestre. Se il job di sostituzione utilizza una codifica dei dati diversa o incompatibile, Dataflow non è in grado di serializzare o deserializzare questi record.

• Rimozione di un'operazione "con stato" dalla pipeline. Se rimuovi le operazioni con stato dalla pipeline, il job sostitutivo potrebbe non superare il controllo di compatibilità. Dataflow può unire più passaggi per maggiore efficienza. Se rimuovi un'operazione dipendente dallo stato da un passaggio unito, il controllo non va a buon fine. Le operazioni con stato includono:

  • Trasformazioni che producono o consumano input aggiuntivi.
  • Letture I/O.
  • Trasformazioni che utilizzano lo stato con chiave.
  • Trasformazioni con unione di finestre.

• Modifica delle variabili DoFn stateful. Per i job di streaming in corso, se la pipeline include DoFn con stato, la modifica delle variabili DoFn con stato potrebbe causare un errore della pipeline.

• Tenta di eseguire il job sostitutivo in una zona geografica diversa. Esegui il job sostitutivo nella stessa zona in cui hai eseguito il job precedente.

Aggiornamento degli schemi

Apache Beam consente ai PCollection di avere schemi con campi denominati, nel qual caso non sono necessari codificatori espliciti. Se i nomi e i tipi di campo di un determinato schema rimangono invariati (inclusi i campi nidificati), lo schema non causa l'errore del controllo dell'aggiornamento. Tuttavia, l'aggiornamento potrebbe essere ancora bloccato se altri segmenti della nuova pipeline non sono compatibili.

Sviluppare gli schemi

Spesso è necessario evolvere lo schema di un PCollection a causa dell'evoluzione delle esigenze aziendali. Il servizio Dataflow consente di apportare le seguenti modifiche a uno schema durante l'aggiornamento della pipeline:

• Aggiunta di uno o più nuovi campi a uno schema, inclusi i campi nidificati.
• Rendere facoltativo (con valori null) un tipo di campo obbligatorio (non null).

La rimozione di campi, la modifica dei nomi dei campi o dei tipi di campo non è consentita durante l'aggiornamento.

Passare dati aggiuntivi a un'operazione ParDo esistente

Puoi passare dati aggiuntivi (out-of-band) a un'operazione ParDo esistente utilizzando uno dei seguenti metodi, a seconda del caso d'uso:

• Serializza le informazioni come campi nella sottoclasse DoFn.
• Tutte le variabili a cui fanno riferimento i metodi in un DoFn anonimo vengono messe in sequenza automaticamente.
• Esegui il calcolo dei dati all'interno di DoFn.startBundle().
• Passa i dati utilizzando ParDo.withSideInputs.

Per ulteriori informazioni, consulta le seguenti pagine:

• Guida alla programmazione di Apache Beam: ParDo, in particolare le sezioni sulla creazione di un DoFn e degli input aggiuntivi.
• Riferimento all'SDK Apache Beam per Java: ParDo