Questa pagina fornisce indicazioni e consigli per eseguire l'upgrade del tuo streaming pipeline di dati. Ad esempio, potresti dover eseguire l'upgrade a una versione più recente SDK Apache Beam o aggiornare il codice della pipeline. Sono disponibili diverse opzioni per adattarsi a scenari diversi.
Le pipeline batch invece si interrompono al completamento del job, le pipeline spesso vengono eseguite in modo continuo per e l'elaborazione dei dati. Pertanto, quando esegui l'upgrade delle pipeline in modalità flusso, devi tenere conto per le seguenti considerazioni:
- Potresti dover ridurre al minimo o evitare le interruzioni della pipeline. In alcuni casi, potresti essere in grado di tollerare un'interruzione temporanea dell'elaborazione durante il deployment di una nuova versione di una pipeline. In altri casi, l'applicazione potrebbe non essere in grado di tollerare interruzioni.
- I processi di aggiornamento della pipeline devono gestire le modifiche allo schema in modo da minimizzare l'interruzione dell'elaborazione dei messaggi e di altri sistemi collegati. Ad esempio, se lo schema dei messaggi in una pipeline di elaborazione degli eventi cambia, potrebbero essere necessarie modifiche dello schema anche nei sink di dati a valle.
Puoi utilizzare uno dei seguenti metodi per aggiornare le pipeline in modalità flusso, a seconda della pipeline e dei requisiti di aggiornamento:
Per ulteriori informazioni sui problemi che potresti riscontrare durante un update e su come evitarli, consulta Convalidare un job sostitutivo e Controllo di compatibilità dei job.
Best practice
- Esegui l'upgrade della versione dell'SDK Apache Beam separatamente da eventuali modifiche al codice della pipeline.
- Testa la pipeline dopo ogni modifica prima di apportare aggiornamenti aggiuntivi.
- Esegui regolarmente l'upgrade della versione dell'SDK Apache Beam utilizzata dalla tua pipeline.
Esegui aggiornamenti in corso
Puoi aggiornare alcune pipeline di streaming in corso senza interrompere il job. Questo scenario è chiamato aggiornamento del job in corso. Gli aggiornamenti dei job in esecuzione sono disponibili solo in circostanze limitate:
- Il job deve utilizzare Streaming Engine.
- Il job deve essere in stato di esecuzione.
- Modificherai solo il numero di worker utilizzati dal job.
Per ulteriori informazioni, consulta Impostare l'intervallo di scalabilità automatica nella pagina Scalabilità automatica orizzontale.
Per istruzioni su come eseguire un aggiornamento di un job in esecuzione, consulta Aggiornare una pipeline esistente.
Avvia un job sostitutivo
Se il job aggiornato è compatibile con quello esistente, puoi aggiornare la pipeline utilizzando l'opzione update. Quando sostituisci un job esistente, un nuovo job esegue il codice della pipeline aggiornato.
Il servizio Dataflow mantiene il nome del job, ma esegue il job di sostituzione 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 il nuovo job
. Per maggiori dettagli, consulta
Effetti della sostituzione di un job.
Dataflow esegue un controllo di compatibilità per assicurarsi che il codice della pipeline aggiornato possa essere implementato in sicurezza nella pipeline in esecuzione. Alcune modifiche al codice causano il fallimento del controllo di compatibilità, ad esempio quando gli input laterali vengono aggiunti o rimossi da un passaggio esistente. Se il controllo di compatibilità non va a buon fine, non puoi eseguire un aggiornamento in situ del job.
Per istruzioni su come avviare un job sostitutivo, consulta Avvia un job sostitutivo.
Se l'aggiornamento della pipeline non è compatibile con il job attuale, devi interrompere e sostituire la pipeline. Se la pipeline non può tollerare i tempi di riposo, esegui pipeline parallele.
Arrestare e sostituire le pipeline
Se puoi interrompere temporaneamente l'elaborazione, puoi annulla o scarica la pipeline, per poi sostituirla con quella aggiornata. L'annullamento di una pipeline fa sì che Dataflow interrompa immediatamente l'elaborazione e arresti le risorse il più rapidamente possibile, il che può causare una perdita di dati in fase di elaborazione, noti come dati in transito. Per evitare perdite di dati, nella maggior parte dei casi è preferibile lo svuotamento. Puoi anche utilizzare gli snapshot di Dataflow per salvare lo stato di una pipeline di inserimento flussi, consentendoti di avviare una nuova versione del job Dataflow senza perdere lo stato. Per ulteriori informazioni, vedi Usa gli snapshot Dataflow.
Lo svuotamento di una pipeline chiude immediatamente qualsiasi processo in corso finestre e attiva tutti attivatori. Anche se i dati in corso non vengono persi, lo svuotamento potrebbe provocare la dati incompleti. In questo caso, le finestre in fase di elaborazione emettono parziali o incomplete che consentono di analizzare i dati e visualizzare i risultati. Per ulteriori informazioni, vedi Effetti dello svuotamento di un job. Al termine del job esistente, avvia un nuovo job in modalità flusso che contenga il codice aggiornato della pipeline, in modo da riprendere l'elaborazione.
Con questo metodo, si verificano tempi di inattività tra il momento in cui il job in modalità flusso esistente viene interrotto e la data e l'ora in cui viene eseguita la pipeline sostitutiva pronti per riprendere l'elaborazione dei dati. Tuttavia, annullare o svuotare una pipeline esistente ed eseguire un nuovo job con la pipeline aggiornata è meno complicato rispetto all'esecuzione di pipeline parallele.
Per istruzioni più dettagliate, consulta Svuotare un job Dataflow. Dopo aver svuotato il job corrente, avvia un nuovo job con lo stesso nome.
Nuovo trattamento dei messaggi con snapshot e ricerca di Pub/Sub
In alcuni casi, dopo aver sostituito o annullato una pipeline svuotata, potresti dover rielaborare i messaggi Pub/Sub consegnati in precedenza. Ad esempio, potresti dover utilizzare una logica aziendale aggiornata per rielaborare i dati. Ricerca Pub/Sub è una funzionalità che consente di ripetere i messaggi da uno snapshot Pub/Sub. Puoi utilizzare la modalità Ricerca Pub/Sub con Dataflow per rielaborare i messaggi inviati al momento della creazione dello snapshot della sottoscrizione.
Durante lo sviluppo e il test, puoi anche usare Pub/Sub Seek per ripeti ripetutamente i messaggi noti per verificare l'output dalla pipeline. Quando utilizzi Pub/Sub Seek, non cercare uno snapshot della sottoscrizione quando l'abbonamento viene utilizzato da una pipeline. In tal caso, la ricerca può invalidare Logica della filigrana di Dataflow e potrebbe influire sul metodo "exactly-once" dei messaggi Pub/Sub.
Un file consigliato gcloud CLI per l'utilizzo di Pub/Sub Seek con Dataflow delle pipeline in una finestra del terminale è il seguente:
Per creare uno snapshot dell'abbonamento, utilizza il comando
gcloud pubsub snapshots create:gcloud pubsub snapshots create SNAPSHOT_NAME --subscription=PIPELINE_SUBSCRIPTION_NAME
Per svuotare o annullare la pipeline, utilizza il comando
gcloud dataflow jobs draino il comandogcloud dataflow jobs cancel:gcloud dataflow jobs drain JOB_ID
o
gcloud dataflow jobs cancel JOB_ID
Per eseguire la ricerca fino allo snapshot, utilizza il comando
gcloud pubsub subscriptions seek:gcloud pubsub subscriptions seek SNAPSHOT_NAME
Esegui il deployment di una nuova pipeline che utilizza l'abbonamento.
Esegui pipeline parallele
Se devi evitare interruzioni della pipeline in modalità flusso durante un aggiornamento, di eseguire pipeline parallele. Crea un nuovo job di streaming con il codice della pipeline aggiornato ed esegui la nuova pipeline in parallelo con quella esistente.
Quando crei la nuova pipeline, utilizza la stessa strategia di definizione delle finestre utilizzata per la pipeline esistente. Lascia in esecuzione la pipeline esistente finché la sua filigrana non supera il timestamp della finestra completa più antica elaborata dalla pipeline aggiornata. Quindi, svuota o annulla la pipeline esistente. La pipeline aggiornata continua a funzionare al suo posto e assume autonomamente il controllo dell'elaborazione.
Il seguente diagramma illustra questa procedura.
Nel diagramma, la pipeline B è il job aggiornato che prende il controllo Pipeline A. Il valore t è il timestamp della finestra completa più antica elaborata dalla pipeline B. Il valore w è la filigrana per la pipeline A. Per semplicità, si assume una filigrana perfetta senza dati in ritardo. L'elaborazione e sono rappresentate sull'asse orizzontale. Entrambe le pipeline utilizzano finestre fisse di cinque minuti (a cascata). I risultati vengono attivati dopo che la filigrana supera la fine di ogni finestra.
Poiché l'output simultaneo si verifica durante il periodo di tempo in cui le due pipeline si sovrappongono, configura le due pipeline in modo da scrivere i risultati in destinazioni diverse. I sistemi a valle possono quindi utilizzare un'astrazione sui due sink di destinazione, ad esempio una vista del database, per eseguire query sui risultati combinati. Questi possono inoltre utilizzare l'astrazione per deduplicare i risultati punto.
L'esempio seguente descrive l'approccio di utilizzo di una pipeline che legge i dati di input da Pub/Sub, esegue alcune elaborazioni e scrive i risultati in BigQuery.
Nello stato iniziale, la pipeline di streaming esistente (Pipeline A) è in esecuzione e legge i messaggi da un argomento Pub/Sub (Topic) utilizzando una sottoscrizione (Subscription A). I risultati sono scritte in una tabella BigQuery (Tabella A). I risultati sono utilizzato tramite una vista BigQuery, che funge da facciata mascherare le modifiche alla tabella sottostante. Questa procedura è un'applicazione di un metodo di progettazione chiamato pattern di facciata. Il seguente diagramma mostra lo stato iniziale.
Crea una nuova sottoscrizione (Sottoscrizione B) per l'aggiornamento una pipeline o un blocco note personalizzato. Esegui il deployment della pipeline aggiornata (Pipeline B), che legge dall'argomento Pub/Sub (Topic) utilizzando la sottoscrizione B e scrive in una tabella BigQuery separata (Tabella B). La il seguente diagramma illustra questo flusso.
A questo punto, la pipeline A e la pipeline B sono in esecuzione in parallelo e scrivere i risultati in tabelle separate. Registriamo il tempo t come Timestamp del primo intervallo di tempo completato elaborato dalla pipeline B.
Quando la filigrana della pipeline A supera il tempo t, svuota la pipeline A. Quando svuota la pipeline, tutte le finestre aperte si chiudono e l'elaborazione per vengono completati i dati mentre sono in corso. Se la pipeline contiene finestre e viene completata sono importanti (supponendo che non siano presenti dati in ritardo), prima di svuotare la pipeline A, lascia che entrambe le pipeline vengano eseguite completamente sovrapposte. Arresta il job di flussi di dati per Pipeline A dopo l'elaborazione e la scrittura di tutti i dati in corso Tabella A. Questo diagramma mostra questa fase.
Al momento è in esecuzione solo la pipeline B. Puoi eseguire query da una visualizzazione BigQuery (visualizzazione facciata), che funge da facciata per Tabella A e Tabella B. Per le righe con lo stesso timestamp in entrambe le tabelle, configura la visualizzazione in modo da restituire le righe della tabella B oppure, se le righe non esistono nella tabella B, utilizza la tabella A come opzione di riserva. Il seguente diagramma mostra la vista (Visualizzazione facciata) che legge sia dalla Tabella A sia dalla Tabella B.
A questo punto, puoi eliminare l'abbonamento A.
Quando vengono rilevati problemi con il deployment di una nuova pipeline, avere pipeline parallele può semplificare il rollback. In questo esempio, potresti mantenere in esecuzione la pipeline A mentre monitori il corretto funzionamento della pipeline B. Se si verificano problemi con la pipeline B, puoi eseguire il rollback Pipeline A.
Limitazioni
Questo approccio presenta i seguenti limiti:
- L'esecuzione di due pipeline sullo stesso input potrebbe generare duplicati i dati nell'output. Il sistema downstream deve essere a conoscenza della tollerare i dati duplicati.
- Quando leggi da un'origine Pub/Sub, l'utilizzo della stessa sottoscrizione per più pipeline non è consigliato e può causare problemi di correttezza. Tuttavia, in alcuni casi d'uso, come pipeline ETL (Extract, Transform, Load), l'utilizzo della stessa sottoscrizione in due pipeline può ridurre i duplicati. In questo scenario sono probabili problemi di scalabilità automatica, ma possono essere limitati mediante la funzionalità di aggiornamento dei lavori in corso. Per ulteriori informazioni, vedi Ottimizza la scalabilità automatica per le pipeline in modalità flusso Pub/Sub.
- Durante la lettura da un'origine Pub/Sub, utilizzando una seconda sottoscrizione genera duplicati, ma non causa problemi di correttezza dei dati. e la scalabilità automatica.
Gestire le mutazioni dello schema
I sistemi di gestione dei dati devono spesso adattarsi alle mutazioni dello schema nel tempo, talvolta a causa di modifiche dei requisiti aziendali e altre volte per motivi tecnici. L'applicazione degli aggiornamenti dello schema in genere richiede un'attenta pianificazione ed esecuzione evitare interruzioni dei sistemi informativi aziendali.
Considera una pipeline che legge i messaggi che contengono payload JSON
di un argomento Pub/Sub. La pipeline converte ogni messaggio in un
TableRow
gcloud e scrive le righe in una tabella BigQuery. Lo schema
della tabella di output è simile ai messaggi elaborati dalla pipeline.
Nel diagramma seguente, lo schema è denominato Schema A.
Nel tempo, lo schema del messaggio potrebbe subire mutazioni in modi non banali. Ad esempio, i campi vengono aggiunti, rimossi o sostituiti. Lo schema A si evolve in un nuovo schema. Nella Nella discussione che segue, il nuovo schema viene chiamato Schema B. Nella in questo caso, la pipeline A deve essere aggiornata e lo schema della tabella di output deve per supportare lo schema B.
Per la tabella di output, puoi eseguire alcune mutazioni dello schema senza downtown.
Ad esempio, puoi aggiungere nuovi campi o
modalità a colonne,
ad esempio da REQUIRED a NULLABLE, senza tempi di inattività.
In genere, queste mutazioni non influiscono sulle query esistenti. Tuttavia, lo schema
le mutazioni che modificano o rimuovono campi dello schema esistenti interrompono le query o i risultati
in altre interruzioni. Il seguente approccio consente di apportare modifiche senza richiedere il tempo di riposo.
Separa i dati scritti dalla pipeline in una tabella principale e in una o più tabelle di staging. La tabella dell'entità archivia i dati storici scritti dalla pipeline. Le tabelle di staging memorizzano l'output più recente della pipeline. Puoi definire Vista facciata di BigQuery sulle tabelle entità e di gestione temporanea che consente ai consumatori di eseguire query su dati storici e aggiornati.
Il seguente diagramma rivede il flusso della pipeline precedente in modo da includere una tabella intermedia (Tabella intermedia A), una tabella principale e una vista di facciata.
Nel flusso rivisto, la pipeline A elabora i messaggi che utilizzano lo schema A e scrive l'output nella Tabella temporanea A, che ha uno schema compatibile. La tabella principale contiene i dati storici scritti dalle versioni precedenti della pipeline, nonché i risultati uniti periodicamente dalla tabella di staging. I consumatori possono eseguire query su dati aggiornati, inclusi dati storici e in tempo reale, mediante la vista delle facciate.
Quando lo schema del messaggio passa da Schema A a Schema B, potresti aggiornare il codice della pipeline in modo che sia compatibile con i messaggi che utilizzano Schema B. La pipeline esistente deve essere aggiornata con la nuova implementazione. Se esegui pipeline parallele, puoi assicurarti che l'elaborazione dei dati in streaming continui senza interruzioni. Terminazione e sostituzione delle pipeline provoca un'interruzione dell'elaborazione, perché nessuna pipeline è in esecuzione per un periodo di nel tempo.
La pipeline aggiornata scrive in un'altra tabella intermedia (Tabella intermedia B) che utilizza lo schema B. Puoi utilizzare un flusso di lavoro orchestrato per creare la nuova tabella di staging prima di aggiornare la pipeline. Aggiorna la vista della facciata per includere i risultati della nuova tabella di staging, eventualmente utilizzando un passaggio del flusso di lavoro correlato.
Il seguente diagramma mostra il flusso aggiornato che mostra la tabella di staging B con lo schema B e come la visualizzazione della facciata viene aggiornata per includere i contenuti della tabella principale e di entrambe le tabelle di staging.
Come procedura separata dall'aggiornamento della pipeline, puoi unire le tabelle intermediarie alla tabella principale, periodicamente o in base alle esigenze. Le seguenti il diagramma mostra come la Tabella temporanea A viene unita nella tabella dell'entità.
Passaggi successivi
- Consulta la procedura dettagliata per aggiornare una pipeline esistente.