Il libro «Kafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale»

Ciao, Habrozhiteli! Questo libro è adatto a qualsiasi sviluppatore che desidera comprendere l'elaborazione dei flussi. Comprendere la programmazione distribuita aiuterà a studiare meglio Kafka e Kafka Streams. Sarebbe utile conoscere anche il framework Kafka stesso, ma non è obbligatorio: ti spiegherò tutto ciò che è necessario. Gli sviluppatori esperti di Kafka, così come i principianti, grazie a questo libro impareranno a creare interessanti applicazioni per l'elaborazione dei flussi utilizzando la libreria Kafka Streams. Gli sviluppatori Java di livello medio e alto, già abituati a concetti come la serializzazione, impareranno a applicare le proprie competenze per creare applicazioni Kafka Streams. Il codice sorgente del libro è scritto in Java 8 e utilizza ampiamente la sintassi delle espressioni lambda di Java 8, quindi saper lavorare con le funzioni lambda (anche in un altro linguaggio di programmazione) ti sarà utile.

Estratto. 5.3. Aggregazione e operazioni a finestre

In questa sezione ci concentreremo sullo studio delle parti più promettenti di Kafka Streams. Fino ad ora abbiamo esaminato i seguenti aspetti di Kafka Streams:

• creazione di una topologia di elaborazione;
• uso dello stato nelle applicazioni di flusso;
• esecuzione di unioni di flussi di dati;
• differenze tra flussi di eventi (KStream) e flussi di aggiornamenti (KTable).

Nei prossimi esempi uniremo tutti questi elementi. Inoltre, ti familiarizzerai con le operazioni a finestre, un'altra fantastica possibilità delle applicazioni di flusso. Il nostro primo esempio sarà una semplice aggregazione.

5.3.1. Aggregazione del volume delle vendite di azioni per settori industriali

L'aggregazione e il raggruppamento sono strumenti essenziali quando si lavora con dati di flusso. Esaminare singoli record man mano che arrivano spesso non è sufficiente. Per estrarre ulteriori informazioni dai dati, è necessario raggrupparli e combinarli.

In questo esempio dovrai indossare il vestito di un trader intraday, che deve monitorare i volumi di vendita delle azioni di aziende in diversi settori industriali. In particolare, ti interessano cinque aziende con i maggiori volumi di vendite di azioni in ciascuno dei settori industriali.

Per tale aggregazione saranno necessari diversi passaggi per trasformare i dati nella forma desiderata (parlando in termini generali).

1. Creare una fonte basata su un argomento che pubblichi informazioni grezze sul commercio di azioni. Dobbiamo convertire un oggetto di tipo StockTransaction in un oggetto di tipo ShareVolume. Il punto è che l'oggetto StockTransaction contiene metadati sulle vendite, mentre a noi servono solo i dati sul numero di azioni vendute.
2. Agruppare i dati ShareVolume per simboli azionari. Dopo aver raggruppato per simboli, possiamo ridurre questi dati a somme intermedie sui volumi di vendita delle azioni. Vale la pena notare che il metodo KStream.groupBy restituisce un'istanza di tipo KGroupedStream. Per ottenere un'istanza di KTable, è necessario chiamare successivamente il metodo KGroupedStream.reduce.

Che cos'è l'interfaccia KGroupedStream

I metodi KStream.groupBy e KStream.groupByKey restituiscono un'istanza di KGroupedStream. KGroupedStream è una rappresentazione intermedia di un flusso di eventi dopo il raggruppamento per chiavi. Non è assolutamente destinato a un utilizzo diretto. Invece, KGroupedStream è utilizzato per operazioni di aggregazione, il cui risultato è sempre un KTable. Poiché il risultato delle operazioni di aggregazione è un KTable e in esse viene applicato uno store di stato, potrebbe non tutti gli aggiornamenti risultanti vengano inviati ulteriormente lungo la pipeline.

Il metodo KTable.groupBy restituisce un KGroupedTable simile: una rappresentazione intermedia di un flusso di aggiornamenti riorganizzati per chiave.

Fermiamoci un attimo e diamo un'occhiata alla figura 5.9, che mostra cosa abbiamo ottenuto. Questa topologia dovrebbe esserti già ben nota.

Diamo ora un'occhiata al codice per questa topologia (che puoi trovare nel file src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (listing 5.2).

Il codice fornito è conciso e compie una grande quantità di operazioni in poche righe. Nel primo parametro del metodo builder.stream puoi notare qualcosa di nuovo per te: il valore dell'enumerazione AutoOffsetReset.EARLIEST (c'è anche LATEST), impostato tramite il metodo Consumed.withOffsetResetPolicy. Questo tipo enumerativo consente di specificare la strategia di ripristino delle offset per ciascun KStream o KTable, avente la precedenza sul parametro di ripristino offset della configurazione.

GroupByKey e GroupBy

Nell'interfaccia KStream ci sono due metodi per raggruppare le registrazioni: GroupByKey e GroupBy. Entrambi restituiscono un KGroupedTable, quindi potresti avere una domanda legittima: qual è la differenza tra di loro e quando usare ciascuno?

Il metodo GroupByKey viene utilizzato quando le chiavi in KStream non sono vuote. E soprattutto, il flag "richiede una nuova partizione" non è mai stato impostato.

Il metodo GroupBy presuppone che tu abbia cambiato le chiavi per la raggruppamento, quindi il flag di ri-partizionamento è impostato su true. Eseguire connessioni, aggregazioni, ecc. dopo il metodo GroupBy porterà a una nuova partizione automatica.
In sintesi: si dovrebbe utilizzare GroupByKey ogni volta che è possibile, piuttosto che GroupBy.

Cosa fanno i metodi mapValues e groupBy è chiaro, quindi diamo un'occhiata al metodo sum() (puoi trovarlo nel file src/main/java/bbejeck/model/ShareVolume.java) (listing 5.3).

Il metodo ShareVolume.sum restituisce la somma intermedia del volume delle azioni, e il risultato dell'intera catena di calcoli rappresenta un oggetto KTable. Ora capisci quale ruolo gioca KTable. Quando gli oggetti ShareVolume arrivano, l'ultimo aggiornamento attuale viene mantenuto nell'oggetto KTable corrispondente. È importante non dimenticare che tutti gli aggiornamenti si riflettono nel precedente shareVolumeKTable, ma non tutti vengono inviati ulteriormente.

Successivamente, utilizziamo questo KTable per eseguire un'aggregazione (in base alla quantità di azioni vendute) per ottenere cinque aziende con i maggiori volumi di vendita di azioni in ciascun settore. Le nostre azioni saranno simili a quelle della prima aggregazione.

1. Eseguire un'altra operazione groupBy per raggruppare i singoli oggetti ShareVolume per settori industriali.
2. Iniziare a sommare gli oggetti ShareVolume. Questa volta l'oggetto di aggregazione rappresenta una coda di priorità di dimensioni fisse. In questa coda di dimensioni fisse vengono mantenute solo cinque aziende con le maggiori quantità di azioni vendute.
3. Mappare le code del punto precedente in un valore stringa e restituire le cinque più vendute per numero di azioni per settori industriali.
4. Scrivere i risultati in formato stringa nel topic.

La figura 5.10 mostra il grafo della topologia del flusso dei dati. Come puoi vedere, il secondo ciclo di elaborazione è piuttosto semplice.

Ora, avendo chiara la struttura di questo secondo ciclo di elaborazione, puoi consultare il suo codice sorgente (lo troverai nel file src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (listing 5.4).

In questo inizializzatore c'è una variabile fixedQueue. È un oggetto personalizzato - un adattatore per java.util.TreeSet, utilizzato per tenere traccia dei N migliori risultati in ordine decrescente della quantità di azioni vendute.

Hai già incontrato le chiamate groupBy e mapValues, quindi non ci soffermeremo su di esse (chiamiamo il metodo KTable.toStream, poiché il metodo KTable.print è considerato obsoleto). Ma non hai ancora visto la versione KTable del metodo aggregate(), quindi dedicheremo un po' di tempo a discuterne.

Come ricordi, KTable si distingue perché le registrazioni con chiavi identiche sono considerate aggiornamenti. KTable sostituisce la vecchia registrazione con quella nuova. L'aggregazione avviene in modo simile: vengono aggregate le ultime registrazioni con una chiave. Quando arriva una registrazione, viene aggiunta all'istanza della classe FixedSizePriorityQueue tramite un sommatorio (il secondo parametro nella chiamata al metodo aggregate), ma se esiste già un'altra registrazione con la stessa chiave, la vecchia registrazione viene rimossa tramite un sottrattore (il terzo parametro nella chiamata al metodo aggregate).

Questo significa che il nostro aggregatore, FixedSizePriorityQueue, non aggrega affatto tutti i valori con una chiave, ma memorizza la somma mobile delle quantità dei N tipi di azioni più venduti. In ogni registrazione in arrivo è contenuto il numero totale di azioni vendute fino a quel momento. KTable ti darà informazioni su quali aziende stanno vendendo il maggior numero di azioni in questo momento, non è richiesta l'aggregazione mobile di ciascun aggiornamento.

Abbiamo imparato a fare due cose importanti:

• raggruppare i valori in KTable in base a una chiave comune;
• eseguire su questi valori raggruppati operazioni utili come la compressione e l'aggregazione.

Essere in grado di eseguire queste operazioni è importante per comprendere il significato dei dati che scorrono attraverso l'applicazione Kafka Streams e scoprire quali informazioni essi portano.

Abbiamo anche unito alcuni dei concetti chiave discussi in precedenza in questo libro. Nel capitolo 4 abbiamo parlato di quanto sia importante per un'applicazione in streaming avere uno stato locale resiliente. Il primo esempio di questo capitolo ha dimostrato perché è così cruciale averlo: consente di tenere traccia delle informazioni che hai già visualizzato. L'accesso locale permette di evitare ritardi di rete, rendendo così l'applicazione più performante e resistente agli errori.

Quando si esegue un'operazione di aggregazione o di riduzione, è necessario specificare il nome dello store di stato. Le operazioni di aggregazione e riduzione restituiscono un'istanza di KTable, e KTable utilizza lo store di stato per sostituire i risultati vecchi con quelli nuovi. Come hai visto, non tutti gli aggiornamenti vengono inviati lungo il pipeline, ed è importante, poiché le operazioni di aggregazione sono destinate a ottenere informazioni finali. Se non si utilizza lo stato locale, KTable invierà avanti tutti i risultati di aggregazione e riduzione.

In seguito esamineremo l'esecuzione di operazioni come l'aggregazione all'interno di un intervallo di tempo specifico, note come operazioni a finestra (windowing operations).

5.3.2. Operazioni a finestra

Nella sezione precedente abbiamo introdotto le operazioni di aggregazione e riduzione "mobile". L'applicazione ha svolto un'aggregazione continua del volume di vendite di azioni seguita dall'aggregazione delle cinque azioni più vendute sul mercato.

A volte sono necessarie operazioni continue di aggregazione e riduzione dei risultati. Altre volte è necessario eseguire operazioni solo su un determinato intervallo di tempo. Ad esempio, calcolare quante transazioni di borsa sono state effettuate per un'azienda specifica negli ultimi 10 minuti. Oppure quante persone hanno cliccato su un nuovo banner pubblicitario negli ultimi 15 minuti. L'applicazione può eseguire tali operazioni ripetutamente, ma con risultati che riguardano solo gli intervalli di tempo specificati (finestre temporali).

Conteggio delle transazioni di borsa per cliente

Nell'esempio seguente ci concentreremo sul monitoraggio delle transazioni di borsa per vari trader — sia grandi organizzazioni che astuti investitori solitari.

Ci sono due possibili motivi per un simile monitoraggio. Uno di questi è la necessità di sapere cosa acquistano/vendono i leader di mercato. Se questi grandi attori e investitori esperti vedono opportunità emergere, ha senso seguire le loro strategie. Il secondo motivo è il desiderio di notare eventuali segni di transazioni illegali che utilizzano informazioni riservate. Per questo, sarà necessario analizzare la correlazione tra grandi picchi nelle vendite e importanti comunicati stampa.

Questo monitoraggio consiste in fasi come:

• creazione di un flusso di lettura dal topic stock-transactions;
• raggruppamento delle registrazioni in arrivo per identificativo dell'acquirente e simbolo di borsa. La chiamata al metodo groupBy restituisce un'istanza della classe KGroupedStream;
• restituzione del flusso di dati limitato a una finestra temporale tramite KGroupedStream.windowedBy, il che consente l'aggregazione su finestre. A seconda del tipo di finestra, viene restituito un TimeWindowedKStream o un SessionWindowedKStream;
• conteggio delle transazioni per l'operazione di aggregazione. Il flusso di dati a finestre determina se una specifica registrazione è considerata in questo conteggio;
• registrazione dei risultati nel topic o stampa in console durante lo sviluppo.

La topologia di questa applicazione è semplice, ma non guasta avere un'immagine chiara di essa. Diamo un'occhiata alla figura 5.11.

Successivamente esamineremo la funzionalità delle operazioni su finestre e il codice corrispondente.

Tipi di finestre

In Kafka Streams esistono tre tipi di finestre:

• finestre di sessione;
• finestre 'tumbling';
• finestre 'sliding'/hopping.

Quale scegliere dipende dai requisiti aziendali. Le finestre 'tumbling' e 'hopping' sono limitate nel tempo, mentre i limiti delle finestre di sessione sono legati alle azioni degli utenti: la durata di ciascuna sessione è determinata esclusivamente dall'attività dell'utente. L'importante è non dimenticare che tutti i tipi di finestre si basano sui timestamp delle registrazioni, e non sull'orario di sistema.

Successivamente implementeremo la nostra topologia con ciascuno dei tipi di finestre. Il codice completo verrà fornito solo nel primo esempio; per gli altri tipi di finestre nulla cambierà, tranne il tipo dell'operazione di finestra.

Finestre di sessione

Le finestre di sessione si differenziano notevolmente da tutti gli altri tipi di finestre. Non sono limitate tanto nel tempo, quanto nell'attività dell'utente (o nell'attività di quell'entità che desideri monitorare). Le finestre di sessione sono delimitate da periodi di inattività.

La figura 5.12 illustra il concetto di finestre di sessione. Una sessione più breve si fonderà con la sessione a sinistra. E la sessione a destra sarà separata, in quanto segue un lungo periodo di inattività. Le finestre di sessione si basano sulle azioni degli utenti, ma utilizzano i timestamp dalle registrazioni per determinare a quale sessione riguarda la registrazione.

Utilizzo delle finestre di sessione per monitorare le transazioni di borsa

Utilizzeremo le finestre di sessione per catturare informazioni sulle transazioni di borsa. L'implementazione delle finestre di sessione è mostrata nel listing 5.5 (che può essere trovato nel file src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java).

La maggior parte delle operazioni di questa topologia le hai già incontrate, quindi non c'è bisogno di rivederle qui. Ma ci sono anche alcuni nuovi elementi che discuteremo ora.

In ogni operazione groupBy viene solitamente eseguita qualche operazione di aggregazione (aggregazione, riduzione o conteggio). È possibile eseguire un'aggregazione cumulativa in modo incrementale, oppure un'aggregazione finestrata, che considera le registrazioni all'interno di una finestra temporale specifica.

Il codice del listing 5.5 esegue il conteggio delle transazioni all'interno delle finestre di sessione. Nella figura 5.13 queste azioni vengono analizzate passo dopo passo.

Con la chiamata windowedBy(SessionWindows.with(twentySeconds).until(fifteenMinutes)) creiamo una finestra di sessione con un intervallo di inattività di 20 secondi e un intervallo di conservazione di 15 minuti. L'intervallo di inattività di 20 secondi significa che l'applicazione includerà qualsiasi registrazione che arriva entro 20 secondi dalla fine o dall'inizio della sessione corrente nella sessione attuale (attiva).

Successivamente, indichiamo quale operazione di aggregazione deve essere eseguita nella finestra di sessione: in questo caso count. Se un record in ingresso esce dai limiti dell'intervallo di inattività (da qualsiasi lato del timestamp), l'applicazione crea una nuova sessione. L'intervallo di conservazione significa mantenere la sessione per un determinato periodo e consente dati tardivi che escono dal periodo di inattività della sessione, ma possono ancora essere collegati. Inoltre, l'inizio e la fine della nuova sessione risultante dalla fusione corrispondono al timestamp più antico e a quello più recente.

Esaminiamo alcuni record dal metodo count per vedere come funzionano le sessioni (tab. 5.1).

All'arrivo dei record, cerchiamo sessioni già esistenti con la stessa chiave, con un orario di fine inferiore all'attuale timestamp — intervallo di inattività — e un orario di inizio superiore all'attuale timestamp + intervallo di inattività. Tenendo conto di questo, quattro record della tab. 5.1 si fondono in un'unica sessione nel seguente modo.

1. Per prima cosa arriva il record 1, quindi l'orario di inizio è uguale a quello di fine ed è 00:00:00.

2. Successivamente arriva il record 2, e cerchiamo sessioni che terminano non prima delle 23:59:55 e iniziano non oltre le 00:00:35. Troviamo il record 1 e uniamo le sessioni 1 e 2. Prendiamo l'orario di inizio della sessione 1 (più antico) e l'orario di fine della sessione 2 (più recente), così la nostra nuova sessione inizia alle 00:00:00 e termina alle 00:00:15.

3. Arriva il record 3, cerchiamo sessioni tra le 00:00:30 e le 00:01:10 e non ne troviamo nessuna. Aggiungiamo la seconda sessione per la chiave 123-345-654,FFBE, che inizia e termina alle 00:00:50.

4. Arriva il record 4, e cerchiamo sessioni tra le 23:59:45 e le 00:00:25. Questa volta troviamo entrambe le sessioni: 1 e 2. Tutte e tre le sessioni si uniscono in una sola, con un orario di inizio alle 00:00:00 e un orario di fine alle 00:00:15.

Dalle informazioni presentate in questa sezione, è importante ricordare i seguenti punti chiave:

• le sessioni non sono finestre di dimensioni fisse. La durata della sessione è determinata dall'attività nell'ambito di un determinato intervallo di tempo;
• i timestamp nei dati determinano se un evento rientra in una sessione esistente o in un periodo di inattività.

Successivamente discuteremo il prossimo tipo di finestre: le finestre 'capovolte'.

'Finestre capovolte'

Le finestre «tumbling» catturano eventi che si verificano in un determinato intervallo di tempo. Immagina di dover registrare tutte le transazioni di borsa di una certa azienda ogni 20 secondi, in modo da raccogliere tutti gli eventi in quel lasso di tempo. Alla fine dell'intervallo di 20 secondi, la finestra «tumbling» si rovescia e passa a un nuovo intervallo di 20 secondi di osservazione. La figura 5.14 illustra questa situazione.

Come puoi vedere, tutti gli eventi ricevuti negli ultimi 20 secondi sono inclusi nella finestra. Al termine di questo intervallo di tempo, viene creata una nuova finestra.

Nel listing 5.6 è fornito il codice che dimostra l'uso delle finestre «tumbling» per catturare ogni 20 secondi le transazioni di borsa (puoi trovarlo nel file src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

Grazie a questa piccola modifica alla chiamata del metodo TimeWindows.of, è possibile utilizzare una finestra «tumbling». In questo esempio non c'è chiamata al metodo until(), quindi verrà utilizzato l'intervallo di conservazione predefinito, che è di 24 ore.

Infine, è ora di passare all'ultima delle opzioni di finestra: le finestre «hopping».

Finestre scorrevoli («hopping»)

Le finestre scorrevoli/«hopping» sono simili alle finestre «tumbling», ma con una piccola differenza. Le finestre scorrevoli non aspettano la fine dell'intervallo di tempo prima di creare una nuova finestra per elaborare eventi recenti. Avviano nuovi calcoli dopo un intervallo di attesa inferiore alla durata della finestra.

Per illustrare le differenze tra finestre «tumbling» e «hopping», torniamo all'esempio del conteggio delle transazioni di borsa. Il nostro obiettivo rimane il conteggio del numero di transazioni, ma non vogliamo attendere l'intero intervallo di tempo prima di aggiornare il contatore. Invece, aggiorneremo il contatore a intervalli di tempo più brevi. Ad esempio, continueremo a contare il numero di transazioni ogni 20 secondi, ma aggiorneremo il contatore ogni 5 secondi, come mostrato nella figura 5.15. In questo modo avremo tre finestre di risultati con dati sovrapposti.

Nel listing 5.7 è fornito il codice per definire finestre scorrevoli (puoi trovarlo nel file src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

La finestra "rotolante" può essere trasformata in una finestra "rimbalzante" aggiungendo una chiamata al metodo advanceBy(). Nell'esempio fornito, l'intervallo di salvataggio è di 15 minuti.

In questa sezione hai visto come limitare i risultati dell'aggregazione con finestre temporali. In particolare, vorrei che ricordassi tre cose importanti da questa sezione:

• la dimensione delle finestre di sessione è limitata non da un intervallo di tempo, ma dall'attività degli utenti;
• le finestre "rotolanti" offrono una panoramica degli eventi all'interno di un determinato periodo di tempo;
• la durata delle finestre "rimbalzanti" è fissa, ma vengono frequentemente aggiornate e possono contenere record sovrapposti in tutte le finestre.

Successivamente scopriremo come riconvertire un KTable in KStream per l'unione.

5.3.3. Unione di oggetti KStream e KTable

Nel capitolo 4 abbiamo discusso dell'unione di due oggetti KStream. Ora dobbiamo imparare a unire KTable e KStream. Questo potrebbe essere necessario per la seguente semplice ragione. KStream è un flusso di record, mentre KTable è un flusso di aggiornamenti, ma a volte può essere necessario aggiungere un contesto supplementare al flusso di record tramite aggiornamenti da KTable.

Prendiamo i dati sul numero di transazioni di borsa e uniamoli alle notizie di borsa relative ai settori industriali pertinenti. Ecco cosa occorre fare per ottenere questo considerando il codice già esistente.

1. Trasformare l'oggetto KTable contenente i dati sulle transazioni di borsa in KStream sostituendo successivamente la chiave con una chiave che indica il settore industriale corrispondente a quel simbolo azionario.
2. Creare un oggetto KTable che legga i dati da un topic con le notizie di borsa. Questo nuovo KTable sarà categorizzato per settori industriali.
3. Unire gli aggiornamenti delle notizie con le informazioni sul numero di transazioni di borsa per i settori industriali.

Ora vediamo come implementare questo piano d'azione.

Trasformazione di KTable in KStream

Per trasformare KTable in KStream, è necessario fare quanto segue.

1. Chiamare il metodo KTable.toStream().
2. Utilizzando la chiamata al metodo KStream.map, sostituire la chiave con il nome del settore industriale, dopodiché estrarre l'oggetto TransactionSummary dall'istanza Windowed.

Collegheremo queste operazioni in una catena nel seguente modo (il codice può essere trovato nel file src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (elenco 5.8).

Poiché eseguiamo l'operazione KStream.map, il ri-partizionamento per l'istanza KStream restituita viene effettuato automaticamente durante il suo utilizzo in una join.

Abbiamo completato il processo di trasformazione, ora dobbiamo creare un oggetto KTable per leggere le notizie di borsa.

Creazione di KTable per le notizie di borsa

Fortunatamente, per creare un oggetto KTable è sufficiente una sola riga di codice (questo codice può essere trovato nel file src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (listing 5.9).

È importante notare che non è necessario specificare oggetti Serde, poiché nelle impostazioni vengono utilizzati Serde per stringhe. Inoltre, grazie all'uso dell'enumerazione EARLIEST, la tabella viene popolata con le registrazioni fin dall'inizio.

Ora possiamo passare all'ultimo passo: la join.

Join tra aggiornamenti delle notizie e dati sul numero di transazioni

Creare una join non presenta difficoltà. Utilizzeremo una join sinistra nel caso in cui non ci siano notizie di borsa per l'industria corrispondente (il codice necessario può essere trovato nel file src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (listing 5.10).

Questo operatore leftJoin è piuttosto semplice. A differenza delle join del Capitolo 4, il metodo JoinWindow non viene utilizzato, poiché durante l'esecuzione della join KStream-KTable esiste solo una registrazione per ogni chiave in KTable. Questa join non è limitata nel tempo: la registrazione è presente in KTable o non lo è affatto. La conclusione principale: mediante gli oggetti KTable è possibile arricchire KStream con dati di riferimento meno frequentemente aggiornati.

Ora esamineremo un metodo più efficiente per arricchire eventi da KStream.

5.3.4. Oggetti GlobalKTable

Come avete capito, c'è la necessità di arricchire i flussi di eventi o di aggiungervi contesto. Nel capitolo 4 avete visto la join tra due oggetti KStream, mentre nella sezione precedente – la join tra KStream e KTable. In tutti questi casi, è necessario un ri-partizionamento del flusso dati quando si mappano le chiavi a un nuovo tipo o valore. A volte il ri-partizionamento avviene esplicitamente, altre volte Kafka Streams lo fa automaticamente. Il ri-partizionamento è necessario poiché le chiavi sono cambiate e le registrazioni devono trovarsi nelle nuove partizioni, altrimenti la join non sarà possibile (questo è stato discusso nel capitolo 4, nella sezione "Ri-partizionamento dei dati" del sotto-capitolo 4.2.4).

La ricostituzione dei segmenti ha un costo

La ricostituzione dei segmenti richiede risorse — costi aggiuntivi per la creazione di argomenti intermedi e la conservazione di dati duplicati in un altro argomento; implica anche un aumento della latenza a causa della scrittura e lettura da questo argomento. Inoltre, se è necessario eseguire un'unione su più di un aspetto o dimensione, è necessario organizzare le unioni in serie, mappare le voci con nuove chiavi e ripetere il processo di ricostituzione dei segmenti.

Unione con set di dati di dimensioni minori

In alcuni casi, il volume dei dati di riferimento con cui si prevede di unire è relativamente ridotto, quindi le loro copie complete possono essere memorizzate localmente su ciascun nodo. Per tali situazioni, Kafka Streams prevede la classe GlobalKTable.

Le istanze di GlobalKTable sono uniche, poiché l'applicazione replica tutti i dati su ciascun nodo. Poiché su ciascun nodo sono presenti tutti i dati, non è necessario segmentare il flusso di eventi in base alla chiave dei dati di riferimento affinché sia accessibile a tutte le sezioni. Con gli oggetti GlobalKTable è possibile eseguire anche unioni senza chiave. Torniamo a uno dei precedenti esempi per dimostrare questa capacità.

Unione di oggetti KStream con oggetti GlobalKTable

Nella sezione 5.3.2 abbiamo eseguito aggregazioni finestrate delle transazioni di borsa per clienti. I risultati di questa aggregazione erano simili ai seguenti:

{customerId='074-09-3705', stockTicker='GUTM'}, 17
{customerId='037-34-5184', stockTicker='CORK'}, 16

Sebbene questi risultati soddisfacessero l'obiettivo fissato, sarebbe stato più conveniente se fosse stato visualizzato anche il nome del cliente e il nome completo dell'azienda. Per aggiungere il nome del cliente e il nome dell'azienda, è possibile eseguire normali unioni, ma sarà necessario effettuare due mappature delle chiavi e ricostituire i segmenti. Con GlobalKTable è possibile evitare costi per tali operazioni.

A tal fine, utilizzeremo l'oggetto countStream dal listing 5.11 (il codice corrispondente può essere trovato nel file src/main/java/bbejeck/chapter_5/GlobalKTableExample.java), unendolo con due oggetti GlobalKTable.

Ne abbiamo già discusso in precedenza, quindi non mi ripeterò. Tuttavia, voglio notare che il codice nella funzione toStream().map è astratto in un oggetto-funzione invece che in un'espressione lambda inline, per una maggiore leggibilità.

Il passo successivo è dichiarare due istanze di GlobalKTable (il codice riportato si trova nel file src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (listing 5.12).

Si noti che i nomi dei topic vengono descritti utilizzando tipi enumerati.

Ora che abbiamo preparato tutti i componenti, resta da scrivere il codice per la connessione (che si trova nel file src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (listing 5.13).

Sebbene in questo codice ci siano due connessioni, sono organizzate in forma di catena, poiché nessuno dei loro risultati viene utilizzato separatamente. I risultati vengono restituiti alla fine dell'intera operazione.

Eseguendo l'operazione di connessione sopra descritta, otterrete risultati del seguente tipo:

{customer='Barney, Smith' company="Exxon", transactions= 17}

Il contenuto è rimasto invariato, ma questi risultati sono più chiari.

Se si considera il capitolo 4, è già stato possibile osservare diversi tipi di connessioni in azione. Sono elencati nella tab. 5.2. Questa tabella riflette le possibilità di connessione attuali per la versione 1.0.0 di Kafka Streams; nelle versioni future potrebbero esserci delle modifiche.

In conclusione, ricordo l'aspetto principale: è possibile connettere flussi di eventi (KStream) e flussi di aggiornamenti (KTable) utilizzando uno stato locale. Inoltre, se la dimensione dei dati di riferimento non è troppo grande, è possibile utilizzare l'oggetto GlobalKTable. Il GlobalKTable replica tutte le partizioni su ciascuno dei nodi dell'applicazione Kafka Streams, garantendo così la disponibilità di tutti i dati, indipendentemente dalla partizione a cui appartiene la chiave.

Vedremo quindi una possibilità di Kafka Streams che permette di osservare le modifiche allo stato senza consumare dati dal topic Kafka.

5.3.5. Stato disponibile per le query

Abbiamo già eseguito diverse operazioni che coinvolgono lo stato e abbiamo sempre visualizzato i risultati nella console (per scopi di sviluppo) o li abbiamo registrati in un topic (per scopi di produzione). Quando si registrano risultati in un topic, è necessario utilizzare un consumatore Kafka per visualizzarli.

La lettura dei dati da questi argomenti può essere considerata una forma di viste materializzate (materialized views). Per le nostre necessità, possiamo utilizzare la definizione di vista materializzata dall'«Wikipedia»: «...oggetto fisico di un database che contiene i risultati di un'istruzione. Ad esempio, può essere una copia locale di dati remoti, oppure un sottoinsieme di righe e/o colonne di una tabella o di risultati di join, o una tabella riepilogativa ottenuta tramite aggregazione» (https://en.wikipedia.org/wiki/Materialized_view).

Kafka Streams consente anche di eseguire query interattive (interactive queries) sugli archivi di stato, il che offre la possibilità di leggere direttamente queste viste materializzate. È importante notare che la query sull'archivio di stato è un'operazione «solo lettura». Grazie a ciò, puoi evitare di preoccuparti di rendere inconsistente lo stato durante l'elaborazione dei dati da parte dell'applicazione.

La capacità di effettuare query dirette sugli archivi di stato è di grande importanza. Ciò significa che è possibile creare applicazioni — cruscotti senza la necessità di prima ottenere i dati dal consumatore Kafka. Aumenta anche l'efficienza dell'applicazione, poiché non è necessario riscrivere nuovamente i dati:

• grazie alla località dei dati, possono essere consultati rapidamente;
• si evita la duplicazione dei dati, poiché non vengono scritti in un archivio esterno.

La cosa principale che vorrei che ricordaste: è possibile eseguire direttamente query sullo stato dall'applicazione. Non si può sottovalutare le opportunità che questo offre. Invece di consumare dati da Kafka e archiviare registrazioni in un database per l'applicazione, è possibile effettuare query sugli archivi di stato con lo stesso risultato. Le query dirette sugli archivi di stato significano una minore quantità di codice (assenza di consumatore) e meno software (non è necessaria una tabella di database per memorizzare i risultati).

Abbiamo coperto una notevole quantità di informazioni in questo capitolo, quindi interrompiamo per un momento la nostra discussione sulle richieste interattive ai magazzini di stato. Ma non preoccupatevi: nel capitolo 9 creeremo un'applicazione semplice: un cruscotto con richieste interattive. Per dimostrare le richieste interattive e le loro possibilità di essere integrate nelle applicazioni Kafka Streams, utilizzeremo alcuni degli esempi di questo e del capitolo precedente.

Riepilogo

• Gli oggetti KStream incarnano flussi di eventi, paragonabili alle inserzioni in un database. Gli oggetti KTable incarnano flussi di aggiornamenti, sono più simili agli aggiornamenti in un database. La dimensione di un oggetto KTable non cresce, le registrazioni vecchie vengono sostituite da nuove.
• Gli oggetti KTable sono necessari per le operazioni di aggregazione.
• Le operazioni di finestra possono suddividere i dati aggregati in intervalli temporali.
• Grazie agli oggetti GlobalKTable, è possibile accedere ai dati di riferimento in qualsiasi punto dell'applicazione, indipendentemente dalla partizione per sezioni.
• È possibile effettuare join tra gli oggetti KStream, KTable e GlobalKTable.

Finora ci siamo concentrati sulla creazione di applicazioni Kafka Streams usando l'API DSL KStream di alto livello. Sebbene l'approccio di alto livello permetta di creare programmi puliti e concisi, il suo utilizzo rappresenta un certo compromesso. Lavorare con il DSL KStream significa aumentare la concisione del codice a fronte di una diminuzione del controllo. Nel capitolo successivo esploreremo l'API a basso livello per i nodi di elaborazione e proveremo altri compromessi. I programmi diventeranno più lunghi di quanto siano stati finora, ma avremo la possibilità di creare praticamente qualsiasi nodo di elaborazione di cui potremmo aver bisogno.

→ Maggiori dettagli sul libro possono essere trovati su sito dell'editore

→ Per gli utenti di Habr, uno sconto del 25% con il coupon — Kafka Streams

→ Dopo il pagamento della versione cartacea del libro, verrà inviata una copia elettronica all'email.

Fonte: habr.com