Questa pagina illustra le best practice da utilizzare durante lo sviluppo delle pipeline Dataflow. L'utilizzo di queste best practice offre i seguenti vantaggi:
- Migliora l'osservabilità e le prestazioni della pipeline
- Maggiore produttività degli sviluppatori
- Migliora la testabilità della pipeline
Gli esempi di codice Apache Beam in questa pagina utilizzano Java, ma i contenuti si applicano agli SDK Apache Beam Java, Python e Go.
Domande da considerare
Quando progetti la tua pipeline, poniti queste domande:
- Dove vengono archiviati i dati di input della pipeline? Quanti set di dati di input avete?
- Che aspetto hanno i tuoi dati?
- Che cosa vuoi fare con i tuoi dati?
- Dove dovrebbero andare i dati di output della pipeline?
- Il job Dataflow utilizza Assured Workloads?
Utilizzare i modelli
Per accelerare lo sviluppo della pipeline, anziché crearne una scrivendo codice Apache Beam, utilizza un modello Dataflow quando possibile. I modelli offrono i seguenti vantaggi:
- I modelli sono riutilizzabili.
- I modelli consentono di personalizzare ciascun job modificando parametri specifici della pipeline.
- Chiunque disponga delle autorizzazioni necessarie può utilizzare il modello per eseguire il deployment della pipeline. Ad esempio, uno sviluppatore può creare un job da un modello e un data scientist nell'organizzazione può eseguire il deployment del modello in un secondo momento.
Puoi utilizzare un modello fornito da Google oppure crearne uno personalizzato. Alcuni modelli forniti da Google ti consentono di aggiungere la logica personalizzata come passaggio della pipeline. Ad esempio, il modello da Pub/Sub a BigQuery fornisce un parametro per eseguire una funzione JavaScript definita dall'utente, archiviata in Cloud Storage.
Poiché i modelli forniti da Google sono open source in base alla licenza Apache 2.0, puoi utilizzarli come base per nuove pipeline. I modelli sono utili anche come esempi di codice. Visualizza il codice del modello nel repository GitHub.
Assured Workloads
Assured Workloads aiuta ad applicare requisiti di sicurezza e conformità per i clienti di Google Cloud. Ad esempio, Regioni e assistenza nell'UE con controlli di sovranità contribuisce ad applicare garanzie di residenza e sovranità dei dati per i clienti nell'UE. Alcune funzionalità di Dataflow possono essere limitate o limitate. Se utilizzi Assured Workloads con Dataflow, tutte le risorse a cui accede la tua pipeline devono trovarsi nel progetto o cartella Assured Workloads dell'organizzazione. Le risorse includono:
- Bucket Cloud Storage
- Set di dati di BigQuery
- Argomenti e sottoscrizioni Pub/Sub
- Set di dati Firestore
- Connettori I/O
In Dataflow, per i job di flussi di dati creati dopo il 7 marzo 2024, tutti i dati degli utenti vengono criptati con CMEK.
Per i job di flussi di dati creati prima del 7 marzo 2024, le chiavi dei dati utilizzate nelle operazioni basate su chiavi, come windowing, raggruppamento e join, non sono protette dalla crittografia CMEK. Per abilitare questa crittografia per i tuoi job, scarica o annulla il job, quindi riavvialo. Per maggiori informazioni, consulta Crittografia degli artefatti dello stato della pipeline.
Condividi dati tra pipeline
Non esiste un meccanismo di comunicazione cross-pipeline specifico di Dataflow per la condivisione dei dati o il contesto di elaborazione tra le pipeline. Puoi utilizzare un'archiviazione durevole come Cloud Storage o una cache in memoria come App Engine per condividere i dati tra le istanze della pipeline.
Pianifica job
Puoi automatizzare l'esecuzione della pipeline nei seguenti modi:
- Usa Cloud Scheduler.
- Utilizza l'operatore Dataflow di Apache Airflow, uno dei numerosi operatori di Google Cloud in un flusso di lavoro di Cloud Composer.
- Esegui processi personalizzati (cron) di job su Compute Engine.
Best practice per la scrittura del codice della pipeline
Le sezioni seguenti forniscono le best practice da utilizzare per la creazione di pipeline mediante la scrittura di codice Apache Beam.
Struttura il codice Apache Beam
Per creare pipeline, è comune utilizzare la trasformazione generica Apache Beam con elaborazione parallela ParDo.
Quando applichi una trasformazione ParDo, fornisci il codice sotto forma di oggetto DoFn. DoFn è una classe SDK Apache Beam che definisce una funzione di elaborazione distribuita.
Puoi considerare il codice DoFn come piccole entità indipendenti: possono
poter esserci molte istanze in esecuzione su macchine diverse, ognuna
senza conoscenza delle altre. Di conseguenza, consigliamo di creare funzioni pure, ideali per la natura parallela e distribuita degli elementi DoFn.
Le funzioni pure hanno le seguenti caratteristiche:
- Le funzioni pure non dipendono dallo stato nascosto o esterno.
- Non hanno effetti collaterali osservabili.
- Sono deterministici.
Il modello di funzione puro non è strettamente rigido. Quando il codice non dipende da cose non garantite dal servizio Dataflow, le informazioni sullo stato o i dati di inizializzazione esterna possono essere validi per DoFn e altri oggetti delle funzioni.
Quando struttura le trasformazioni di ParDo e crei i tuoi elementi DoFn,
tieni presente le seguenti linee guida:
- Quando utilizzi l'elaborazione "exactly-once", il servizio Dataflow garantisce che ogni elemento nell'input
PCollectionvenga elaborato da un'istanzaDoFnesattamente una volta. - Il servizio Dataflow non garantisce quante volte viene richiamato un
DoFn. - Il servizio Dataflow non garantisce esattamente come sono raggruppati gli elementi distribuiti. Non garantisce quali elementi, se presenti, vengono elaborati insieme.
- Il servizio Dataflow non garantisce il numero esatto di
DoFnistanze create nel corso di una pipeline. - Il servizio Dataflow è a tolleranza di errore e potrebbe riprovare il codice più volte se i worker riscontrano problemi.
- Il servizio Dataflow potrebbe creare copie di backup del tuo codice. Potrebbero verificarsi problemi con gli effetti collaterali manuali, ad esempio se il codice si basa su file temporanei con nomi non univoci o se crea file temporanei.
- Il servizio Dataflow serializza l'elaborazione degli elementi per
DoFnistanza. Il codice non deve essere rigorosamente sicuro per i thread, ma qualsiasi stato condiviso tra più istanzeDoFndeve essere sicuro per i thread.
Crea librerie di trasformazioni riutilizzabili
Il modello di programmazione Apache Beam consente di riutilizzare le trasformazioni. Creando una libreria condivisa di trasformazioni comuni, puoi migliorare la riutilizzabilità, la testabilità e la proprietà del codice da parte di diversi team.
Considera i due seguenti esempi di codice Java, che leggono entrambi gli eventi di pagamento. Supponendo che entrambe le pipeline eseguano la stessa elaborazione, possono utilizzare le stesse trasformazioni tramite una libreria condivisa per le restanti fasi di elaborazione.
Il primo esempio proviene da un'origine Pub/Sub illimitata:
PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options)
// Initial read transform
PCollection<PaymentEvent> payments =
p.apply("Read from topic",
PubSubIO.readStrings().withTimestampAttribute(...).fromTopic(...))
.apply("Parse strings into payment events",
ParDo.of(new ParsePaymentEventFn()));
Il secondo esempio proviene da un'origine di database relazionale vincolata:
PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);
PCollection<PaymentEvent> payments =
p.apply(
"Read from database table",
JdbcIO.<PaymentEvent>read()
.withDataSourceConfiguration(...)
.withQuery(...)
.withRowMapper(new RowMapper<PaymentEvent>() {
...
}));
Il modo in cui implementi le best practice per la riusabilità del codice varia a seconda del linguaggio di programmazione e dello strumento di creazione. Ad esempio, se utilizzi Maven, puoi separare il codice di trasformazione in un modulo a sé stante. Puoi quindi includere il modulo come sottomodulo in progetti multimodulo più ampi per diverse pipeline, come mostrato nell'esempio di codice seguente:
// Reuse transforms across both pipelines
payments
.apply("ValidatePayments", new PaymentTransforms.ValidatePayments(...))
.apply("ProcessPayments", new PaymentTransforms.ProcessPayments(...))
...
Per ulteriori informazioni, consulta le seguenti pagine della documentazione di Apache Beam:
- Requisiti per la scrittura del codice utente per le trasformazioni Apache Beam
- Guida di stile per
PTransform: una guida di stile per gli autori di nuove raccoltePTransformriutilizzabili
Utilizzare le code di messaggi non recapitabili per la gestione degli errori
A volte la pipeline non è in grado di elaborare gli elementi. Una causa comune sono i problemi relativi ai dati. Ad esempio, un elemento contenente JSON non formattato correttamente può causare errori di analisi.
Sebbene sia possibile rilevare eccezioni all'interno del metodo DoFn.ProcessElement, registrare l'errore e eliminare l'elemento, questo approccio perde i dati e impedisce che i dati vengano ispezionati in un secondo momento per la gestione manuale o la risoluzione dei problemi.
Utilizza invece un pattern chiamato coda messaggi non recapitabili (coda di messaggi non elaborati).
Rileva le eccezioni nel metodo DoFn.ProcessElement e gli errori di log. Anziché eliminare l'elemento con errori, utilizza gli output di diramazione per scrivere gli elementi con errori in un oggetto PCollection separato. Questi elementi vengono quindi scritti in un sink di dati per ispezionarli e gestirli in un secondo momento con una trasformazione separata.
Il seguente esempio di codice Java mostra come implementare il pattern della coda dei messaggi non recapitabili.
TupleTag<Output> successTag = new TupleTag<>() {};
TupleTag<Input> deadLetterTag = new TupleTag<>() {};
PCollection<Input> input = /* ... */;
PCollectionTuple outputTuple =
input.apply(ParDo.of(new DoFn<Input, Output>() {
@Override
void processElement(ProcessContext c) {
try {
c.output(process(c.element()));
} catch (Exception e) {
LOG.severe("Failed to process input {} -- adding to dead-letter file",
c.element(), e);
c.sideOutput(deadLetterTag, c.element());
}
}).withOutputTags(successTag, TupleTagList.of(deadLetterTag)));
// Write the dead-letter inputs to a BigQuery table for later analysis
outputTuple.get(deadLetterTag)
.apply(BigQueryIO.write(...));
// Retrieve the successful elements...
PCollection<Output> success = outputTuple.get(successTag);
// and continue processing ...
Utilizza Cloud Monitoring per applicare criteri di monitoraggio e avviso diversi per la coda dei messaggi non recapitabili della pipeline. Ad esempio, puoi visualizzare il numero e le dimensioni degli elementi elaborati dalla trasformazione dei messaggi non recapitabili e configurare gli avvisi in modo che vengano attivati se vengono soddisfatte determinate condizioni di soglia.
Gestire le mutazioni dello schema
Puoi gestire dati che hanno schemi imprevisti ma validi utilizzando un pattern di messaggi non recapitabili, che scrive gli elementi con errori in un oggetto PCollection separato.
In alcuni casi, vuoi gestire automaticamente gli elementi
che riflettono uno schema mutato come elementi validi. Ad esempio, se lo schema di un elemento riflette una mutazione come l'aggiunta di nuovi campi, puoi adattare lo schema del sink di dati per adattarlo alle mutazioni.
La mutazione automatica dello schema si basa sull'approccio di ramificazione-output utilizzato dal pattern dei messaggi non recapitabili. Tuttavia, in questo caso attiva una trasformazione che muta lo schema di destinazione ogni volta che vengono riscontrati schemi additivi. Per un esempio di questo approccio, consulta How to handle mutating JSON schemas in a streaming pipeline, with Square Enix sul blog di Google Cloud.
Decidi come unire i database
L'unione di set di dati è un caso d'uso comune per le pipeline di dati. Puoi utilizzare
gli input aggiuntivi o la trasformazione CoGroupByKey per eseguire i join nella pipeline.
Ognuna ha vantaggi e svantaggi.
Gli input aggiuntivi forniscono un modo flessibile per risolvere problemi comuni di elaborazione dei dati, come l'arricchimento dei dati e le ricerche con chiave. A differenza degli oggetti PCollection, gli input aggiuntivi sono
mutabili e possono essere determinati in fase di runtime. Ad esempio, i valori in un input laterale potrebbero essere calcolati da un altro ramo nella pipeline o determinati richiamando un servizio remoto.
Dataflow supporta gli input aggiuntivi inserendo i dati in modo permanente in un'archiviazione permanente, come in un disco condiviso. Questa configurazione rende l'input aggiuntivo disponibile per tutti i worker.
Tuttavia, le dimensioni di input laterale possono essere molto grandi e potrebbero non rientrare nella memoria dei worker. La lettura da un input laterale di grandi dimensioni può causare problemi di prestazioni se i worker devono leggere costantemente dall'archiviazione permanente.
La trasformazione CoGroupByKey
è una
trasformazione Apache Beam principale
che unisce (appiattisce) più oggetti PCollection e raggruppa elementi che hanno una chiave comune. A differenza di un input aggiuntivo, che rende disponibili tutti i dati di input aggiuntivo
per ciascun worker, CoGroupByKey esegue un'operazione di shuffling (raggruppamento)
per distribuire i dati tra i worker. CoGroupByKey è quindi ideale quando gli
PCollection oggetti che vuoi unire sono molto grandi e non rientrano nella memoria
dei worker.
Segui queste linee guida per decidere se utilizzare gli input aggiuntivi o CoGroupByKey:
- Utilizza gli input laterali quando uno degli oggetti
PCollectionche stai unendo è sproporzionatamente più piccolo degli altri e l'oggettoPCollectionpiù piccolo rientra nella memoria del worker. La memorizzazione nella cache dell'input aggiuntivo nella memoria rende il recupero degli elementi rapido ed efficiente. - Utilizza gli input aggiuntivi quando hai un oggetto
PCollectionche deve essere unito più volte nella pipeline. Anziché utilizzare più trasformazioniCoGroupByKey, crea un singolo input lato che può essere riutilizzato da più trasformazioniParDo. - Utilizza
CoGroupByKeyse devi recuperare una grande proporzione di un oggettoPCollectionche supera in modo significativo la memoria dei worker.
Per ulteriori informazioni, consulta Risolvere i problemi di esaurimento della memoria di Dataflow.
Riduci al minimo le costose operazioni per elemento
Un'istanza DoFn elabora batch di elementi chiamati
gruppi,
che sono unità atomiche di lavoro composte da zero o più
elementi. I singoli elementi vengono quindi elaborati con il metodo DoFn.ProcessElement, che viene eseguito per tutti gli elementi. Poiché il metodo DoFn.ProcessElement
viene chiamato per ogni elemento, tutte le operazioni dispendiose in termini di tempo o di calcolo
richiamate da quel metodo vengono eseguite per ogni singolo elemento elaborato dal metodo.
Se devi eseguire operazioni costose solo una volta per un gruppo di elementi,
includele nel metodo DoFn.Setup o nel metodo DoFn.StartBundle
anziché nell'elemento DoFn.ProcessElement. Ecco alcuni esempi:
Analizzare un file di configurazione che controlla alcuni aspetti del comportamento dell'istanza
DoFn. Richiama questa azione solo una volta, quando l'istanzaDoFnviene inizializzata, utilizzando il metodoDoFn.Setup.Creare un'istanza per un client di breve durata che viene riutilizzato in tutti gli elementi di un bundle, ad esempio quando tutti gli elementi del bundle vengono inviati tramite un'unica connessione di rete. Richiama questa azione una volta per set utilizzando il metodo
DoFn.StartBundle.
Limita le dimensioni dei batch e le chiamate simultanee a servizi esterni
Quando chiami servizi esterni, puoi ridurre i costi generali per chiamata utilizzando la trasformazione GroupIntoBatches. Questa trasformazione crea batch di elementi di una dimensione specificata.
Il batch invia elementi a un servizio esterno come un singolo payload anziché individualmente.
In combinazione con il raggruppamento in batch, limita il numero massimo di chiamate parallele (simultanee) al servizio esterno scegliendo le chiavi appropriate per partizionare i dati in entrata. Il numero di partizioni determina la parallelizzazione massima. Ad esempio, se a ogni elemento viene assegnata la stessa chiave, una trasformazione downstream per la chiamata al servizio esterno non viene eseguita in parallelo.
Prendi in considerazione uno dei seguenti approcci per produrre chiavi per gli elementi:
- Scegli un attributo del set di dati da utilizzare come chiavi dei dati, ad esempio gli ID utente.
- Genera chiavi di dati per suddividere gli elementi in modo casuale in un numero fisso di partizioni, dove il numero di possibili coppie chiave-valore determina il numero di partizioni. Devi creare un numero sufficiente di partizioni per il parallelismo.
Ogni partizione deve avere elementi sufficienti affinché la trasformazione
GroupIntoBatchessia utile.
Il seguente esempio di codice Java mostra come suddividere in modo casuale gli elementi in dieci partizioni:
// PII or classified data which needs redaction.
PCollection<String> sensitiveData = ...;
int numPartitions = 10; // Number of parallel batches to create.
PCollection<KV<Long, Iterable<String>>> batchedData =
sensitiveData
.apply("Assign data into partitions",
ParDo.of(new DoFn<String, KV<Long, String>>() {
Random random = new Random();
@ProcessElement
public void assignRandomPartition(ProcessContext context) {
context.output(
KV.of(randomPartitionNumber(), context.element()));
}
private static int randomPartitionNumber() {
return random.nextInt(numPartitions);
}
}))
.apply("Create batches of sensitive data",
GroupIntoBatches.<Long, String>ofSize(100L));
// Use batched sensitive data to fully utilize Redaction API,
// which has a rate limit but allows large payloads.
batchedData
.apply("Call Redaction API in batches", callRedactionApiOnBatch());
Identifica i problemi di rendimento causati da passaggi combinati
Dataflow crea un grafico dei passaggi che rappresenta la pipeline in base alle trasformazioni e ai dati che hai usato per realizzarla. Questo grafico è chiamato grafico di esecuzione della pipeline.
Quando esegui il deployment della pipeline, Dataflow
potrebbe modificare il grafico di esecuzione della pipeline. Ad esempio, Dataflow potrebbe fondere alcune operazioni, un processo noto come ottimizzazione della fusione, per evitare l'impatto in termini di prestazioni e costi della scrittura di ogni oggetto PCollection intermedio nella pipeline.
In alcuni casi, Dataflow potrebbe determinare erroneamente il modo ottimale per fondere le operazioni nella pipeline, il che può limitare la capacità del job di utilizzare tutti i worker disponibili. In questi casi, è possibile impedire l'unione delle operazioni.
Considera il seguente esempio di codice Apache Beam. Una trasformazione GenerateSequence crea un piccolo oggetto PCollection limitato, che viene poi ulteriormente elaborato da due trasformazioni ParDo a valle.
La trasformazione Find Primes Less-than-N potrebbe essere costosa dal punto di vista del calcolo e
è probabile che venga eseguita lentamente per grandi numeri. Al contrario, la trasformazione Increment Number probabilmente viene completata rapidamente.
import com.google.common.math.LongMath;
...
public class FusedStepsPipeline {
final class FindLowerPrimesFn extends DoFn<Long, String> {
@ProcessElement
public void processElement(ProcessContext c) {
Long n = c.element();
if (n > 1) {
for (long i = 2; i < n; i++) {
if (LongMath.isPrime(i)) {
c.output(Long.toString(i));
}
}
}
}
}
public static void main(String[] args) {
Pipeline p = Pipeline.create(options);
PCollection<Long> sequence = p.apply("Generate Sequence",
GenerateSequence
.from(0)
.to(1000000));
// Pipeline branch 1
sequence.apply("Find Primes Less-than-N",
ParDo.of(new FindLowerPrimesFn()));
// Pipeline branch 2
sequence.apply("Increment Number",
MapElements.via(new SimpleFunction<Long, Long>() {
public Long apply(Long n) {
return ++n;
}
}));
p.run().waitUntilFinish();
}
}
Il seguente diagramma mostra una rappresentazione grafica della pipeline nell'interfaccia di monitoraggio di Dataflow.
L'interfaccia di monitoraggio di Dataflow mostra che si verifica la stessa lenta velocità di elaborazione per entrambe le trasformazioni, in particolare 13 elementi al secondo. Ci si potrebbe aspettare che la trasformazione Increment Number elabori
gli elementi rapidamente, ma che invece sembra essere associata alla stessa velocità
di elaborazione di Find Primes Less-than-N.
Il motivo è che Dataflow ha unito i passaggi in un'unica
fase, che ne impedisce l'esecuzione in modo indipendente. Puoi utilizzare il comando gcloud dataflow jobs describe per trovare ulteriori informazioni:
gcloud dataflow jobs describe --full job-id --format json
Nell'output risultante, i passaggi fusi sono descritti nell'oggetto ExecutionStageSummary nell'array di ComponentTransform:
...
"executionPipelineStage": [
{
"componentSource": [
...
],
"componentTransform": [
{
"name": "s1",
"originalTransform": "Generate Sequence/Read(BoundedCountingSource)",
"userName": "Generate Sequence/Read(BoundedCountingSource)"
},
{
"name": "s2",
"originalTransform": "Find Primes Less-than-N",
"userName": "Find Primes Less-than-N"
},
{
"name": "s3",
"originalTransform": "Increment Number/Map",
"userName": "Increment Number/Map"
}
],
"id": "S01",
"kind": "PAR_DO_KIND",
"name": "F0"
}
...
In questo scenario, poiché la trasformazione Find Primes Less-than-N è il passaggio lento,
interrompere la fusione prima di questo passaggio è una strategia appropriata. Un metodo per annullare i passaggi è inserire una trasformazione GroupByKey e separarla prima del passaggio, come mostrato nel seguente esempio di codice Java.
sequence
.apply("Map Elements", MapElements.via(new SimpleFunction<Long, KV<Long, Void>>() {
public KV<Long, Void> apply(Long n) {
return KV.of(n, null);
}
}))
.apply("Group By Key", GroupByKey.<Long, Void>create())
.apply("Emit Keys", Keys.<Long>create())
.apply("Find Primes Less-than-N", ParDo.of(new FindLowerPrimesFn()));
Puoi anche combinare questi passaggi di annullamento della fusione in una trasformazione composita riutilizzabile.
Dopo aver annullato la fusione dei passaggi, quando esegui la pipeline, Increment Number
viene completato in pochi secondi e la trasformazione Find Primes Less-than-N
molto più lunga viene eseguita in una fase separata.
In questo esempio viene applicata un'operazione di raggruppamento e di separazione dei passaggi per annullare l'unione dei passaggi.
Per altre circostanze, puoi avvalerti di altri approcci. In questo caso, la gestione dell'output duplicato non è un problema, dato l'output consecutivo della trasformazione GenerateSequence.
Gli oggetti KV
con chiavi duplicate vengono deduplicati in una singola chiave nella trasformazione del gruppo (GroupByKey) e della trasformazione di separazione (Keys). Per mantenere i duplicati dopo le operazioni di gruppo e di separazione,
crea coppie chiave-valore seguendo questa procedura:
- Utilizza una chiave casuale e l'input originale come valore.
- Raggruppare utilizzando la chiave casuale.
- Emetti i valori per ogni chiave come output.
Puoi anche utilizzare una trasformazione Reshuffle per evitare la fusione delle trasformazioni circostanti. Tuttavia, gli effetti collaterali della trasformazione Reshuffle non sono trasferibile tra diversi runner Apache Beam.
Per maggiori informazioni sull'ottimizzazione del parallelismo e della fusione, consulta Ciclo di vita della pipeline.
Usa le metriche Apache Beam per raccogliere insight sulle pipeline
Le metriche Apache Beam sono una classe di utilità che produce metriche per generare report sulle proprietà di una pipeline in esecuzione. Quando utilizzi Cloud Monitoring, le metriche Apache Beam sono disponibili come metriche personalizzate di Cloud Monitoring.
L'esempio seguente mostra le metriche Counter di Apache Beam utilizzate in una sottoclasse DoFn.
Il codice di esempio utilizza due contatori. Un contatore monitora gli errori di analisi JSON (malformedCounter), mentre l'altro controlla se il messaggio JSON è valido, ma contiene un payload vuoto (emptyCounter). In Cloud Monitoring, i nomi delle metrica personalizzata sono custom.googleapis.com/dataflow/malformedJson e custom.googleapis.com/dataflow/emptyPayload. Puoi usare le metriche personalizzate
per creare criteri di visualizzazione e avviso in Cloud Monitoring.
final TupleTag<String> errorTag = new TupleTag<String>(){};
final TupleTag<MockObject> successTag = new TupleTag<MockObject>(){};
final class ParseEventFn extends DoFn<String, MyObject> {
private final Counter malformedCounter = Metrics.counter(ParseEventFn.class, "malformedJson");
private final Counter emptyCounter = Metrics.counter(ParseEventFn.class, "emptyPayload");
private Gson gsonParser;
@Setup
public setup() {
gsonParser = new Gson();
}
@ProcessElement
public void processElement(ProcessContext c) {
try {
MyObject myObj = gsonParser.fromJson(c.element(), MyObject.class);
if (myObj.getPayload() != null) {
// Output the element if non-empty payload
c.output(successTag, myObj);
}
else {
// Increment empty payload counter
emptyCounter.inc();
}
}
catch (JsonParseException e) {
// Increment malformed JSON counter
malformedCounter.inc();
// Output the element to dead-letter queue
c.output(errorTag, c.element());
}
}
}
Scopri di più
Le pagine seguenti forniscono ulteriori informazioni su come strutturare la pipeline, come scegliere quali trasformazioni applicare ai dati e cosa considerare quando scegli i metodi di input e output della pipeline.
Per scoprire di più sulla creazione del codice utente, consulta i requisiti per le funzioni fornite dall'utente.