Aggiornare una pipeline esistente

L'aggiornamento delle pipeline batch non è supportato.

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

• Vuoi migliorare o perfezionare 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 a Container-Optimized OS per tutti i worker Dataflow.
• Vuoi scalare una pipeline Apache Beam in modalità flusso per utilizzare un numero diverso di worker.

Puoi aggiornare i job in due modi:

• Aggiornamento dei job in corso: per i job di flussi di dati 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 della pipeline aggiornato o per aggiornare le opzioni del job non supportate dagli aggiornamenti dei job in corso, avvia un nuovo job che sostituisce quello esistente. Per verificare se un job di sostituzione è valido, prima di avviare il nuovo job, convalida il relativo grafico del job.

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

Puoi anche utilizzare l'infrastruttura di logging integrata dell'SDK Apache Beam per registrare le informazioni quando aggiorni il job. Per saperne di più, consulta Utilizza i log della pipeline. Per identificare i problemi relativi al codice della pipeline, utilizza il livello di logging DEBUG.

• Per istruzioni sull'aggiornamento dei job di flussi di dati che utilizzano modelli classici, consulta Aggiorna un job di flussi di dati con un modello personalizzato.
• Per istruzioni sull'aggiornamento dei job di flussi di dati che utilizzano i template flessibili, segui le istruzioni di gcloud CLI in questa pagina o consulta Aggiorna un job con template flessibile.

Aggiornamento delle opzioni dei job in corso

Per un job in modalità flusso che utilizza Streaming Engine, puoi aggiornare le seguenti opzioni del job senza interromperlo 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: l'utilizzo della CPU target, nell'intervallo [0.1, 0.9]
• latency-tier: il livello di latenza, che imposta la soglia di backlog. Valori: low_latency (predefinito, meno di 15 secondi), medium_latency (meno di 2 minuti) o high_latency (meno di 10 minuti).

Per altri aggiornamenti dei job, devi sostituire il job corrente con quello aggiornato. Per saperne di più, consulta Avvia un job di sostituzione.

Esegui un aggiornamento in corso

Per eseguire un aggiornamento delle opzioni del job in corso, segui questi passaggi.

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 \
  --worker-utilization-hint=TARGET_UTILIZATION \
  --latency-tier=TIER \
  JOB_ID

Sostituisci quanto segue:

• REGION: l'ID della regione del job
• MINIMUM_WORKERS: il numero minimo di istanze di Compute Engine
• MAXIMUM_WORKERS: il numero massimo di istanze Compute Engine
• TARGET_UTILIZATION: un valore compreso nell'intervallo [0.1, 0.9]
• TIER: il livello di latenza target. Uno dei seguenti valori: low_latency, medium_latency o high_latency
• JOB_ID: l'ID del job da aggiornare

Puoi anche aggiornare --min-num-workers, --max-num-workers, --worker-utilization-hint o --latency-tier singolarmente.

REST

Utilizza il metodo projects.locations.jobs.update:

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,
    "latency_tier": TIER
  }
}

Sostituisci quanto segue:

• MASK: un elenco separato da virgole di parametri da aggiornare, tra i seguenti:
  • runtime_updatable_params.max_num_workers
  • runtime_updatable_params.min_num_workers
  • runtime_updatable_params.worker_utilization_hint
• PROJECT_ID: l'ID progetto Google Cloud del job Dataflow
• REGION: l'ID della regione 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
• TARGET_UTILIZATION: un valore compreso nell'intervallo [0.1, 0.9]
• TIER: il livello di latenza target. Uno dei seguenti: low_latency, medium_latency o high_latency

Puoi anche aggiornare min_num_workers, max_num_workers, worker_utilization_hint o latency_tier 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
  }
}

Un job deve essere in stato di esecuzione per essere idoneo agli aggiornamenti in corso. Se il job non è stato avviato o è già stato annullato, si verifica un errore. Analogamente, se avvii un job di sostituzione, attendi che inizi l'esecuzione 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 di sostituzione è valido, prima di avviarlo, convalida il relativo grafico del job. In Dataflow, un grafico del job è una rappresentazione grafica di una pipeline. Convalidando il grafico del job, riduci il rischio che la pipeline riscontri errori o errori della pipeline dopo l'aggiornamento. Inoltre, puoi convalidare gli aggiornamenti senza dover interrompere il job originale, in modo che non si verifichino tempi di inattività.

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

Java

• Passa l'opzione --update.
• Imposta l'opzione --jobName in PipelineOptions sullo stesso nome del job che vuoi aggiornare.
• Imposta l'opzione --region sulla stessa regione del job che vuoi aggiornare.
• Includi l'opzione di servizio --dataflowServiceOptions=graph_validate_only.
• Se i nomi delle trasformazioni nella pipeline sono cambiati, devi fornire una mappatura delle trasformazioni e passarla utilizzando l'opzione --transformNameMapping.
• Se invii un job sostitutivo che utilizza una versione successiva dell'SDK Apache Beam, imposta --updateCompatibilityVersion sulla versione dell'SDK Apache Beam utilizzata nel job originale.

Python

• Passa l'opzione --update.
• Imposta l'opzione --job_name in PipelineOptions sullo stesso nome del job che vuoi aggiornare.
• Imposta l'opzione --region sulla stessa regione del job che vuoi aggiornare.
• Includi l'opzione di servizio --dataflow_service_options=graph_validate_only.
• Se i nomi delle trasformazioni nella pipeline sono cambiati, devi fornire una mappatura delle trasformazioni e passarla utilizzando l'opzione --transform_name_mapping.
• Se invii un job sostitutivo che utilizza una versione successiva dell'SDK Apache Beam, imposta --updateCompatibilityVersion sulla versione dell'SDK Apache Beam utilizzata nel job originale.

Vai

• Passa l'opzione --update.
• Imposta l'opzione --job_name con lo stesso nome del job che vuoi aggiornare.
• Imposta l'opzione --region sulla stessa regione del job che vuoi aggiornare.
• Includi l'opzione di servizio --dataflow_service_options=graph_validate_only.
• Se i nomi delle trasformazioni nella pipeline sono cambiati, devi fornire una mappatura delle trasformazioni e passarla utilizzando l'opzione --transform_name_mapping.

gcloud

Per convalidare il grafico del job per un job con template flessibile, utilizza il comando gcloud dataflow flex-template run con l'opzione additional-experiments:

• Passa l'opzione --update.
• Imposta JOB_NAME sullo stesso nome del job che vuoi aggiornare.
• Imposta l'opzione --region sulla stessa regione del job che vuoi aggiornare.
• Includi l'opzione --additional-experiments=graph_validate_only.
• Se i nomi delle trasformazioni nella pipeline sono cambiati, devi fornire una mappatura delle trasformazioni e passarla utilizzando l'opzione --transform-name-mappings.

Ad esempio:

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

Sostituisci JOB_NAME con il nome del job che vuoi aggiornare.

REST

Utilizza il campo additionalExperiments nell'oggetto FlexTemplateRuntimeEnvironment (template flessibili) o RuntimeEnvironment.

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

L'opzione di servizio graph_validate_only convalida solo gli aggiornamenti della pipeline. Non utilizzare questa opzione durante la creazione o l'avvio delle pipeline. Per aggiornare la pipeline, avvia un job di sostituzione 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 del 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à. Il contenuto del messaggio dipende dall'errore.

Avvia un job di sostituzione

Potresti sostituire un job esistente per i seguenti motivi:

• Per eseguire il codice della pipeline aggiornato.
• Per aggiornare le opzioni dei job che non supportano gli aggiornamenti in corso.

Per verificare se un job di sostituzione è valido, prima di avviare il nuovo job, convalida il relativo grafico del 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:

Java

• Passa l'opzione --update.
• Imposta l'opzione --jobName in PipelineOptions sullo stesso nome del job che vuoi aggiornare.
• Imposta l'opzione --region sulla stessa regione del job che vuoi aggiornare.
• Se i nomi delle trasformazioni nella pipeline sono cambiati, devi fornire una mappatura delle trasformazioni e passarla utilizzando l'opzione --transformNameMapping.
• Se invii un job sostitutivo che utilizza una versione successiva dell'SDK Apache Beam, imposta --updateCompatibilityVersion sulla versione dell'SDK Apache Beam utilizzata nel job originale.

Python

• Passa l'opzione --update.
• Imposta l'opzione --job_name in PipelineOptions sullo stesso nome del job che vuoi aggiornare.
• Imposta l'opzione --region sulla stessa regione del job che vuoi aggiornare.
• Se i nomi delle trasformazioni nella pipeline sono cambiati, devi fornire una mappatura delle trasformazioni e passarla utilizzando l'opzione --transform_name_mapping.
• Se invii un job sostitutivo che utilizza una versione successiva dell'SDK Apache Beam, imposta --updateCompatibilityVersion sulla versione dell'SDK Apache Beam utilizzata nel job originale.

Vai

• Passa l'opzione --update.
• Imposta l'opzione --job_name con lo stesso nome del job che vuoi aggiornare.
• Imposta l'opzione --region sulla stessa regione del job che vuoi aggiornare.
• Se i nomi delle trasformazioni nella pipeline sono cambiati, devi fornire una mappatura delle trasformazioni e passarla utilizzando l'opzione --transform_name_mapping.

gcloud

Per aggiornare un job con template flessibile utilizzando gcloud CLI, utilizza il comando gcloud dataflow flex-template run. L'aggiornamento di altri job tramite gcloud CLI non è supportato.

• Passa l'opzione --update.
• Imposta JOB_NAME sullo stesso nome del job che vuoi aggiornare.
• Imposta l'opzione --region sulla stessa regione del job che vuoi aggiornare.
• Se i nomi delle trasformazioni nella pipeline sono cambiati, devi fornire una mappatura delle trasformazioni e passarla utilizzando l'opzione --transform-name-mappings.

REST

Queste istruzioni mostrano come aggiornare i job non basati su modelli utilizzando l'API REST. Per utilizzare l'API REST per aggiornare un job con modello classico, consulta Aggiorna un job di streaming con modello personalizzato. Per utilizzare l'API REST per aggiornare un job con template flessibile, consulta Aggiorna un job con template flessibile.

1. Recupera la risorsa job per il job che vuoi sostituire utilizzando il metodo projects.locations.jobs.get. Includi il parametro di query view con il valore JOB_VIEW_DESCRIPTION. L'inclusione di JOB_VIEW_DESCRIPTION limita la quantità di dati nella risposta, in modo che la richiesta successiva non superi i limiti di dimensione. Se hai bisogno di informazioni più dettagliate sul job, utilizza il valore JOB_VIEW_ALL.

   GET https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID?view=JOB_VIEW_DESCRIPTION
   

   Sostituisci i seguenti valori:

   • PROJECT_ID: l'ID progetto Google Cloud del job Dataflow
   • REGION: la regione del job che vuoi aggiornare
   • JOB_ID: l'ID job del job che vuoi aggiornare

2. Per aggiornare il job, utilizza il metodo projects.locations.jobs.create. Nel corpo della richiesta, utilizza la risorsa job che hai recuperato.

   POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs
   {
     "id": JOB_ID,
     "replaceJobId": JOB_ID,
     "name": JOB_NAME,
     "type": "JOB_TYPE_STREAMING",
     "transformNameMapping": {
       string: string,
       ...
     },
   }
   

   Sostituisci quanto segue:

   • JOB_ID: lo stesso ID job dell'ID del job che vuoi aggiornare.
   • JOB_NAME: lo stesso nome del job che vuoi aggiornare.

   Se i nomi delle trasformazioni nella pipeline sono cambiati, devi fornire una mappatura delle trasformazioni e passarla utilizzando il campo transformNameMapping.

3. (Facoltativo) Per inviare la richiesta utilizzando curl (Linux, macOS o Cloud Shell), salva la richiesta in un file JSON, quindi esegui questo comando:

   curl -X POST -d "@FILE_PATH" -H "Content-Type: application/json" -H "Authorization: Bearer $(gcloud auth print-access-token)"  https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs
   

   Sostituisci FILE_PATH con il percorso del file JSON che contiene il corpo della richiesta.

Attenzione: per facilitare la mappatura tra le trasformazioni nella pipeline precedente e nella pipeline sostitutiva, assegna nomi espliciti a ogni trasformazione nelle pipeline. Per impostazione predefinita, Dataflow genera un avviso se non assegni un nome esplicito alle trasformazioni. Puoi aumentare questo avviso impostandolo come errore.

Specifica il nome del job sostitutivo

Java

Quando avvii il job di sostituzione, il valore che passi per l'opzione --jobName deve corrispondere esattamente al nome del job che vuoi sostituire.

Python

Quando avvii il job di sostituzione, il valore che passi per l'opzione --job_name deve corrispondere esattamente al nome del job che vuoi sostituire.

Vai

Quando avvii il job di sostituzione, il valore che passi per l'opzione --job_name deve corrispondere esattamente al nome del job che vuoi sostituire.

gcloud

Quando avvii il job sostitutivo, JOB_NAME deve corrispondere esattamente al nome del job che vuoi sostituire.

REST

Imposta il valore del campo replaceJobId sullo stesso ID job del job che vuoi aggiornare. Per trovare il valore corretto del nome del job, seleziona il job precedente nell'interfaccia di monitoraggio di Dataflow. Poi, nel riquadro laterale Informazioni job, individua il campo ID job.

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

Figura 1: il riquadro laterale Informazioni job per un job Dataflow in esecuzione con il campo Nome job.

In alternativa, esegui una query su un elenco di job esistenti utilizzando l'interfaccia a riga di comando di Dataflow. Inserisci il comando gcloud dataflow jobs list nella finestra della shell o del terminale per ottenere un elenco dei job Dataflow nel tuo progetto Google Cloude trova 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

Crea una mappatura di trasformazione

Se la pipeline sostitutiva 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 sostitutiva.

Java

Passa la mappatura utilizzando l'opzione della riga di comando --transformNameMapping, utilizzando il seguente formato generale:

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

Devi fornire solo le voci di mappatura in --transformNameMapping per i nomi delle trasformazioni che sono cambiati tra la pipeline precedente e quella sostitutiva.

Quando esegui il comando con --transformNameMapping, potresti dover utilizzare i caratteri di escape per le virgolette, a seconda della shell. Ad esempio, in Bash:

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

Python

Passa la mappatura utilizzando l'opzione della riga di comando --transform_name_mapping, utilizzando il seguente formato generale:

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

Devi fornire solo le voci di mappatura in --transform_name_mapping per i nomi delle trasformazioni che sono cambiati tra la pipeline precedente e quella sostitutiva.

Quando esegui con --transform_name_mapping, potresti dover utilizzare i caratteri di escape per le virgolette, a seconda della shell. Ad esempio, in Bash:

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

Vai

Passa la mappatura utilizzando l'opzione della riga di comando --transform_name_mapping, utilizzando il seguente formato generale:

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

Devi fornire solo le voci di mappatura in --transform_name_mapping per i nomi delle trasformazioni che sono cambiati tra la pipeline precedente e quella sostitutiva.

Quando esegui il comando con --transform_name_mapping, potresti dover utilizzare i caratteri di escape per le virgolette, a seconda della shell. Ad esempio, in Bash:

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

gcloud

Passa la mappatura utilizzando l'opzione --transform-name-mappings, con il seguente formato generale:

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

Devi fornire solo le voci di mappatura in --transform-name-mappings per i nomi delle trasformazioni che sono cambiati tra la pipeline precedente e quella sostitutiva.

Quando esegui il comando con --transform-name-mappings, potresti dover utilizzare i caratteri di escape per le virgolette, a seconda della shell. Ad esempio, in Bash:

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

REST

Passa la mappatura utilizzando il campo transformNameMapping, con il seguente formato generale:

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

Devi fornire solo le voci di mappatura in transformNameMapping per i nomi delle trasformazioni che sono cambiati tra la pipeline precedente e quella sostitutiva.

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:

Java

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

Python

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

Vai

  // 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 grafico di esecuzione del job nell'interfaccia di monitoraggio di Dataflow:

Figura 2: il grafico di esecuzione di una pipeline WordCount, come mostrato nell'interfaccia di monitoraggio di Dataflow.

Denominazione delle trasformazioni composite

I nomi delle trasformazioni sono gerarchici e si basano sulla gerarchia delle trasformazioni nella pipeline. Se la pipeline ha una trasformazione composita, le trasformazioni nidificate vengono denominate in base alla trasformazione che le contiene. 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.

Rifattorizza la gerarchia di trasformazione

Se la pipeline sostitutiva utilizza una gerarchia di trasformazione diversa rispetto alla pipeline precedente, devi dichiarare esplicitamente la mappatura. Potresti avere una gerarchia di trasformazioni diversa perché hai rifattorizzato le trasformazioni composite o la pipeline dipende da una trasformazione composita di una libreria che è stata modificata.

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

Java

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

Se elimini completamente una trasformazione nella pipeline sostitutiva, devi fornire una mappatura null. Supponiamo che la pipeline sostitutiva rimuova completamente la trasformazione CountWidgets/Parse:

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

Python

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

Se elimini completamente una trasformazione nella pipeline sostitutiva, devi fornire una mappatura null. Supponiamo che la pipeline sostitutiva rimuova completamente la trasformazione CountWidgets/Parse:

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

Vai

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

Se elimini completamente una trasformazione nella pipeline sostitutiva, devi fornire una mappatura null. Supponiamo che la pipeline sostitutiva rimuova completamente la trasformazione CountWidgets/Parse:

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

gcloud

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

Se elimini completamente una trasformazione nella pipeline sostitutiva, devi fornire una mappatura null. Supponiamo che la pipeline sostitutiva rimuova completamente la trasformazione CountWidgets/Parse:

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

REST

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

Se elimini completamente una trasformazione nella pipeline sostitutiva, devi fornire una mappatura null. Supponiamo che la pipeline sostitutiva rimuova completamente la trasformazione CountWidgets/Parse:

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

Gli 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 sostitutivo con un ID job aggiornato. Questo processo potrebbe causare tempi di inattività durante l'arresto del job esistente, l'esecuzione del controllo di compatibilità e l'avvio del nuovo job.

Il job di sostituzione conserva i seguenti elementi:

• Dati di stato intermedi del job precedente. Le cache in memoria non vengono salvate.
• Record di dati o metadati attualmente "in transito" nel buffer dal job precedente. Ad esempio, alcuni record della pipeline potrebbero essere memorizzati nel buffer in attesa della risoluzione di una finestra.
• Aggiornamenti delle opzioni del job in corso che hai applicato al job precedente.

Se la pipeline contiene connettori I/O gestito, Dataflow verifica la presenza di aggiornamenti ai connettori e utilizza automaticamente l'ultima versione valida nota. Per saperne di più, consulta Upgrade automatici

Dati di stato intermedi

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 memoria cache durante l'aggiornamento della pipeline, come soluzione alternativa, rifattorizza la pipeline per convertire le cache in dati di stato o in input secondari. Per ulteriori informazioni sull'utilizzo degli input secondari, consulta Pattern di input secondari nella documentazione di Apache Beam.

Le pipeline di streaming hanno limiti di dimensioni per ValueState e per gli input secondari. 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 corso di pubblicazione

I dati "in transito" vengono comunque elaborati dalle trasformazioni nella nuova pipeline. Tuttavia, le trasformazioni aggiuntive che aggiungi nel codice della pipeline sostitutiva potrebbero avere o meno effetto, a seconda di dove vengono memorizzati i record nel buffer. In questo esempio, la pipeline esistente ha le seguenti trasformazioni:

Java

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

Python

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

Vai

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

Puoi sostituire il job con il nuovo codice della pipeline nel seguente modo:

Java

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

Python

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

Vai

  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" trasferite dal job precedente.

Modifica la finestra temporale

Puoi modificare le strategie di windowing e trigger per gli elementi PCollection nella pipeline di sostituzione, ma fai attenzione. La modifica delle strategie di windowing o di trigger non influisce sui dati già memorizzati nel buffer o in transito.

Ti consigliamo di apportare solo modifiche minori al windowing della pipeline, ad esempio modificando la durata delle finestre temporali fisse o scorrevoli. Apportare modifiche importanti al windowing o ai trigger, ad esempio cambiare l'algoritmo di windowing, potrebbe avere risultati imprevedibili sull'output della pipeline.

Verifica della compatibilità del 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à viene superato, il job precedente viene interrotto. Il job sostitutivo viene quindi avviato sul servizio Dataflow, mantenendo lo stesso nome. Se il controllo di compatibilità non va a buon fine, il job precedente continua a essere eseguito sul servizio Dataflow e il job sostitutivo restituisce un errore.

Java

A causa di una limitazione, devi utilizzare l'esecuzione con blocco per visualizzare gli errori dei tentativi di aggiornamento non riusciti nella console o nel terminale. L'attuale soluzione alternativa prevede i seguenti passaggi:

1. Utilizza pipeline.run().waitUntilFinish() nel codice della pipeline.
2. Esegui il programma della pipeline di sostituzione con l'opzione --update.
3. Attendi che il job di sostituzione superi il controllo di compatibilità.
4. Esci dal processo runner di blocco digitando Ctrl+C.

In alternativa, puoi monitorare lo stato del job di sostituzione nell'interfaccia di monitoraggio di Dataflow. Se il job è stato avviato correttamente, ha superato anche il controllo di compatibilità.

Python

A causa di una limitazione, devi utilizzare l'esecuzione con blocco per visualizzare gli errori dei tentativi di aggiornamento non riusciti nella console o nel terminale. La soluzione alternativa attuale prevede i seguenti passaggi:

1. Utilizza pipeline.run().wait_until_finish() nel codice della pipeline.
2. Esegui il programma della pipeline di sostituzione con l'opzione --update.
3. Attendi che il job di sostituzione superi il controllo di compatibilità.
4. Esci dal processo runner di blocco digitando Ctrl+C.

In alternativa, puoi monitorare lo stato del job di sostituzione nell'interfaccia di monitoraggio di Dataflow. Se il job è stato avviato correttamente, ha superato anche il controllo di compatibilità.

Vai

A causa di una limitazione, devi utilizzare l'esecuzione con blocco per visualizzare gli errori dei tentativi di aggiornamento non riusciti nella console o nel terminale. In particolare, devi specificare l'esecuzione non bloccante utilizzando i flag --execute_async o --async. L'attuale soluzione alternativa prevede i seguenti passaggi:

1. Esegui il programma della pipeline di sostituzione con l'opzione --update e senza i flag --execute_async o --async.
2. Attendi che il job di sostituzione superi il controllo di compatibilità.
3. Esci dal processo runner di blocco digitando Ctrl+C.

gcloud

A causa di una limitazione, devi utilizzare l'esecuzione con blocco per visualizzare gli errori dei tentativi di aggiornamento non riusciti nella console o nel terminale. La soluzione alternativa attuale prevede i seguenti passaggi:

1. Per le pipeline Java, utilizza pipeline.run().waitUntilFinish() nel codice della pipeline. Per le pipeline Python, utilizza pipeline.run().wait_until_finish() nel codice della pipeline. Per le pipeline Go, segui i passaggi nella scheda Go.
2. Esegui il programma della pipeline di sostituzione con l'opzione --update.
3. Attendi che il job di sostituzione superi il controllo di compatibilità.
4. Esci dal processo runner di blocco digitando Ctrl+C.

REST

A causa di una limitazione, devi utilizzare l'esecuzione con blocco per visualizzare gli errori dei tentativi di aggiornamento non riusciti nella console o nel terminale. La soluzione alternativa attuale prevede i seguenti passaggi:

• Per le pipeline Java, utilizza pipeline.run().waitUntilFinish() nel codice della pipeline. Per le pipeline Python, utilizza pipeline.run().wait_until_finish() nel codice della pipeline. Per le pipeline Go, segui i passaggi nella scheda Go.
• Esegui il programma della pipeline sostitutiva con il campo replaceJobId.
• Attendi che il job di sostituzione superi il controllo di compatibilità.
• Esci dal processo runner di blocco digitando Ctrl+C.

Il controllo di compatibilità utilizza la mappatura di trasformazione 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 i PCollection nella pipeline utilizzino gli stessi Coder. La modifica di un Coder può causare il fallimento del controllo di compatibilità perché i dati in transito o i record memorizzati nel buffer potrebbero non essere serializzati correttamente nella pipeline sostitutiva.

Evita interruzioni della compatibilità

Alcune differenze tra la pipeline precedente e quella sostitutiva possono causare l'esito negativo 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. Questo processo di corrispondenza aiuta Dataflow a trasferire i dati dello stato intermedio per ogni passaggio. Se rinomini o rimuovi dei passaggi, devi fornire una mappatura di trasformazione in modo che Dataflow possa abbinare i dati di stato di conseguenza.
• Modifica degli input secondari per un passaggio. L'aggiunta o la rimozione di input secondari da una trasformazione nella pipeline sostitutiva causa l'esito negativo del controllo di compatibilità.
• Modifica del programmatore per un passaggio. Quando aggiorni un job, Dataflow conserva tutti i record di dati attualmente nel buffer e li gestisce nel job sostitutivo. Ad esempio, i dati memorizzati nel buffer potrebbero verificarsi durante la risoluzione del windowing. Se il job sostitutivo utilizza una codifica dei dati diversa o incompatibile, Dataflow non è in grado di serializzare o deserializzare questi record.

Attenzione: il servizio Dataflow non può garantire che la modifica di un coder nella pipeline precedente con un coder incompatibile causi l'esito negativo del controllo di compatibilità. Ti consigliamo di non tentare di apportare modifiche incompatibili con le versioni precedenti a Coder quando aggiorni la pipeline. Se l'aggiornamento della pipeline riesce, ma riscontri problemi o errori nei dati risultanti, assicurati che la pipeline sostitutiva utilizzi una codifica dei dati compatibile con il job precedente.

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

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

• Modifica delle variabili stateful DoFn. Per i job di flussi di dati in corso, se la pipeline include DoFn stateful, la modifica delle variabili DoFn stateful potrebbe causare l'errore della pipeline.

• Tentativo di eseguire il job di sostituzione in un'altra zona geografica. Esegui il job di sostituzione nella stessa zona in cui hai eseguito il job precedente, altrimenti il job di sostituzione non andrà a buon fine.

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 per un determinato schema rimangono invariati (inclusi i campi nidificati), lo schema non causa l'esito negativo del controllo dell'aggiornamento. Tuttavia, l'aggiornamento potrebbe comunque essere bloccato se altri segmenti della nuova pipeline sono incompatibili.

Evolvi gli schemi

Spesso è necessario evolvere lo schema di un PCollection a causa dell'evoluzione dei requisiti 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 (nullable) un tipo di campo obbligatorio (non nullable).

Durante l'aggiornamento non è consentito rimuovere campi, modificare i nomi dei campi o i tipi di campo.

Passare dati aggiuntivi a un'operazione ParDo esistente

Puoi passare dati aggiuntivi (fuori banda) 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 serializzate automaticamente.
• Calcola i dati all'interno di DoFn.startBundle().
• Passa i dati utilizzando ParDo.withSideInputs.

Per saperne di più, consulta le pagine seguenti:

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