Ciao, Habrozhiteli! Questo libro è adatto a qualsiasi sviluppatore che desidera comprendere l'elaborazione dei flussi. Capire la programmazione distribuita aiuterà a studiare meglio Kafka e Kafka Streams. Sarebbe utile conoscere anche il framework Kafka, ma non è obbligatorio: vi racconterò tutto ciò che serve. Gli sviluppatori esperti di Kafka, così come i neofiti, grazie a questo libro apprenderanno come creare interessanti applicazioni per l'elaborazione dei flussi utilizzando la libreria Kafka Streams. Gli sviluppatori Java di livello medio e alto, già familiari con concetti come la serializzazione, impareranno ad 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 la capacità di lavorare con le funzioni lambda (anche in un altro linguaggio di programmazione) sarà utile.
Estratto. 5.3. Operazioni di aggregazione e finestre
In questo capitolo passeremo a esplorare le parti più promettenti di Kafka Streams. Fino a questo momento abbiamo esaminato i seguenti aspetti di Kafka Streams:
- creazione della topologia di elaborazione;
- utilizzo dello stato nelle applicazioni in streaming;
- esecuzione di connessioni di flussi di dati;
- differenze tra flussi di eventi (KStream) e flussi di aggiornamenti (KTable).
Nei seguenti esempi mettere insieme tutti questi elementi. Inoltre, vi presenteremo le operazioni a finestra, un'altra straordinaria funzionalità delle applicazioni in streaming. Il nostro primo esempio riguarderà una semplice aggregazione.
5.3.1. Aggregazione del volume delle vendite delle azioni per settori industriali
L'aggregazione e la raggruppamento sono strumenti vitali quando si lavora con dati in streaming. Esaminare singoli record man mano che arrivano spesso si rivela insufficiente. Per estrarre informazioni aggiuntive dai dati è necessaria la loro aggregazione e combinazione.
In questo esempio dovrete indossare il costume di un trader intraday, che deve monitorare i volumi delle vendite delle azioni di società in vari settori industriali. In particolare, vi interessano cinque società con i maggiori volumi di vendita di azioni in ciascun settore.
Per ottenere un simile aggregato, saranno necessari diversi passaggi per convertire i dati nella forma desiderata (parlando in termini generali).
- Creare una fonte basata su un argomento che pubblichi informazioni grezze sul trading di azioni. Dobbiamo mappare un oggetto di tipo StockTransaction in un oggetto di tipo ShareVolume. Il fatto è che l'oggetto StockTransaction contiene metadati sulle vendite, mentre noi abbiamo bisogno solo dei dati relativi al numero di azioni vendute.
- Raggruppare i dati ShareVolume per simboli azionari. Dopo aver raggruppato per simboli, è possibile comprimere questi dati in somme intermedie dei volumi di vendita delle azioni. È importante notare che il metodo KStream.groupBy restituisce un'istanza di tipo KGroupedStream. Per ottenere un'istanza di KTable, è possibile chiamare successivamente il metodo KGroupedStream.reduce.
Cos'è l'interfaccia KGroupedStream
I metodi KStream.groupBy e KStream.groupByKey restituiscono un'istanza di KGroupedStream. KGroupedStream rappresenta in modo intermedio un flusso di eventi dopo la raggruppamento per chiavi. Non è destinato ad essere utilizzato direttamente. Invece, KGroupedStream è utilizzato per operazioni di aggregazione, il cui risultato è sempre un KTable. E poiché le operazioni di aggregazione producono un KTable e utilizzano uno store di stato, è possibile che non tutti gli aggiornamenti vengano inviati ulteriormente attraverso la pipeline.
Il metodo KTable.groupBy restituisce un KGroupedTable simile, un rappresentazione intermedia di un flusso di aggiornamenti raggruppati per chiave.
Facciamo una breve pausa e osserviamo la figura 5.9, in cui è mostrato cosa abbiamo ottenuto. Questa topologia dovrebbe esserti già familiare.
Diamo un'occhiata al codice per questa topologia (si trova nel file src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (elenco 5.2).
Il codice fornito si distingue per la sua brevità e per l'ampio numero di operazioni eseguite in poche righe. Nel primo parametro del metodo builder.stream puoi notare qualcosa di nuovo per te: il valore dell'enumerazione AutoOffsetReset.EARLIEST (esiste anche LATEST), impostato tramite il metodo Consumed.withOffsetResetPolicy. Utilizzando questo tipo enumerativo, puoi specificare la strategia di reset degli offset per ogni KStream o KTable, che ha priorità sul parametro di reset degli offset nella configurazione.
GroupByKey e GroupBy
Nell'interfaccia KStream ci sono due metodi per raggruppare le registrazioni: GroupByKey e GroupBy. Entrambi restituiscono un KGroupedTable, quindi potrebbe sorgere un legittimo quesito: qual è la differenza tra di loro e quando utilizzare l'uno o l'altro?
Il metodo GroupByKey si applica quando le chiavi nel KStream sono già non vuote. Inoltre, il flag 'richiesta di ri-partizionamento' non è mai stato impostato.
Il metodo GroupBy presuppone che tu abbia modificato le chiavi per la raggruppamento, quindi il flag di ri-partizionamento è impostato su true. L'esecuzione dopo il metodo GroupBy di join, aggregazioni, ecc. comporterà un ri-partizionamento automatico.
Riassunto: si dovrebbe usare GroupByKey ogni volta che è possibile, invece di GroupBy.
Cosa fanno i metodi mapValues e groupBy è chiaro, quindi diamo un'occhiata al metodo sum() (che si trova 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 capite il ruolo di KTable. Quando gli oggetti ShareVolume arrivano, l'ultima aggiornamento rilevante viene mantenuto nell'oggetto KTable corrispondente. È importante non dimenticare che tutti gli aggiornamenti sono riflessi nel precedente shareVolumeKTable, ma non tutti vengono inoltrati ulteriormente.
Successivamente, utilizziamo questo KTable per aggregare (in base al numero di azioni vendute) per ottenere cinque aziende con i maggiori volumi di vendita di azioni in ciascun settore industriale. Le nostre azioni saranno simili a quelle della prima aggregazione.
- Eseguire un'altra operazione di groupBy per raggruppare i singoli oggetti ShareVolume in base ai settori industriali.
- Iniziare a sommare gli oggetti ShareVolume. Questa volta, l'oggetto di aggregazione è una coda a priorità di dimensione fissa. In questa coda di dimensione fissa vengono mantenute solo cinque aziende con le maggiori quantità di azioni vendute.
- Mostrare le code dal punto precedente come valore stringa e restituire le cinque più vendute per numero di azioni suddivise per settori industriali.
- Registrare i risultati in forma di stringa nel topic.
Nella figura 5.10 è mostrato il grafo della topologia del flusso dei dati. Come potete vedere, il secondo cerchio di elaborazione è piuttosto semplice.
Ora che avete chiaramente compreso la struttura di questo secondo cerchio di elaborazione, potete rivolgervi al suo codice sorgente (lo troverete nel file src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (elenco 5.4).
In questo inizializzatore c'è una variabile fixedQueue. Questo è un oggetto utente - 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 sfide di groupBy e mapValues, quindi non ci fermeremo 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 alla sua discussione.
Come ricordi, la caratteristica di KTable è che le registrazioni con la stessa chiave sono considerate aggiornamenti. KTable sostituisce la vecchia registrazione con la nuova. L'aggregazione avviene in modo simile: vengono aggregate le ultime registrazioni con una stessa chiave. Quando un record arriva, viene aggiunto all'istanza della classe FixedSizePriorityQueue tramite un sommatorie (il secondo parametro nella chiamata del 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 del metodo aggregate).
Questo significa che il nostro aggregatore, FixedSizePriorityQueue, non aggrega tutti i valori con una chiave unica, ma conserva la somma mobile delle quantità delle N azioni più vendute. In ogni record in arrivo è contenuto il numero totale di azioni vendute fino a quel momento. KTable ti fornirà informazioni su quali azioni di quale azienda sono attualmente le più vendute, non è necessaria un'aggregazione mobile di ogni aggiornamento.
Abbiamo imparato a fare due cose importanti:
- raggruppare i valori in KTable per una chiave comune;
- eseguire su questi valori raggruppati operazioni utili come la riduzione e l'aggregazione.
Essere in grado di eseguire queste operazioni è fondamentale per comprendere il significato dei dati che si muovono attraverso l'applicazione Kafka Streams e scoprire quali informazioni portano.
Abbiamo anche riunito alcuni dei concetti chiave discussi in precedenza in questo libro. Nel capitolo 4 abbiamo parlato dell'importanza dello stato locale a prova di guasto per le applicazioni in streaming. Il primo esempio di questo capitolo ha dimostrato perché lo stato locale sia così fondamentale: consente di tenere traccia delle informazioni che hai già visualizzato. L'accesso locale evita ritardi di rete, rendendo l'applicazione più performante e resistente agli errori.
Quando si esegue qualsiasi operazione di aggregazione o riduzione, è necessario specificare il nome dello stato di archiviazione. Le operazioni di aggregazione e riduzione restituiscono un'istanza di KTable, mentre KTable utilizza lo stato di archiviazione per sostituire i vecchi risultati con i nuovi. Come hai visto, non tutti gli aggiornamenti vengono inviati ulteriormente nel pipeline, e questo è importante, poiché le operazioni di aggregazione sono progettate per fornire informazioni finali. Se non si utilizza lo stato locale, KTable invierà ulteriormente tutti i risultati di aggregazione e riduzione.
In seguito, analizzeremo l'esecuzione di operazioni come l'aggregazione all'interno di un intervallo di tempo specifico — le cosiddette operazioni a finestra (windowing operations).
5.3.2. Operazioni a finestra
Nella sezione precedente abbiamo introdotto la convoluzione e l'aggregazione 'mobile'. L'applicazione ha effettuato una convoluzione continua del volume delle vendite delle azioni seguita dall'aggregazione delle cinque azioni più vendute in borsa.
A volte, tali aggregazioni e convoluzioni continue sono necessarie. Altre volte, è necessario eseguire operazioni solo su un intervallo di tempo specifico. Ad esempio, calcolare quante operazioni di borsa sono state effettuate su un'azione di una particolare azienda negli ultimi 10 minuti. Oppure quanti utenti hanno cliccato su un nuovo banner pubblicitario negli ultimi 15 minuti. L'applicazione può eseguire tali operazioni più volte, ma con risultati relativi solo agli intervalli di tempo specificati (finestre temporali).
Conteggio delle transazioni di borsa per acquirente
Nel seguente esempio, ci occuperemo del monitoraggio delle transazioni di borsa tra diversi trader—sia grandi aziende che astuti finanzieri solitari.
Ci sono due possibili motivi per tale monitoraggio. Uno è la necessità di sapere cosa acquistano/vendono i leader di mercato. Se questi grandi attori e investitori esperti vedono delle opportunità, ha senso seguire la loro strategia. Il secondo motivo è il desiderio di notare eventuali segni di negoziazioni illecite che utilizzano informazioni riservate. Per fare ciò, è necessario analizzare la correlazione tra grandi picchi di vendite e importanti comunicati stampa.
Questo monitoraggio è composto da fasi quali:
- creazione di un flusso di lettura dal topic stock-transactions;
- raggruppamento delle registrazioni in ingresso per identificativo del compratore e simbolo del titolo azionario. La chiamata al metodo groupBy restituisce un'istanza della classe KGroupedStream;
- Il ritorno tramite il metodo KGroupedStream.windowedBy del flusso di dati limitato da una finestra temporale, che consente di eseguire aggregazioni a finestra. A seconda del tipo di finestra, viene restituito un TimeWindowedKStream o un SessionWindowedKStream;
- il conteggio delle transazioni per l'operazione di aggregazione. Il flusso di dati a finestra determina se un particolare record viene incluso in questo conteggio;
- la registrazione dei risultati in un topic o la loro visualizzazione nella console durante lo sviluppo.
La topologia di questa applicazione è semplice, ma un'immagine esplicativa non guasterebbe. Diamo un'occhiata alla fig. 5.11.
In seguito, esamineremo la funzionalità delle operazioni a finestra e il codice corrispondente.
Tipi di finestre
In Kafka Streams ci sono tre tipi di finestre:
- finestra di sessione;
- finestra a scorrimento (tumbling);
- finestra mobile / a scatti (sliding / hopping).
La scelta dipende dalle esigenze aziendali. Le finestre "rimbalzanti" e "saltellanti" sono limitate nel tempo, mentre i limiti delle finestre di sessione sono legati alle azioni degli utenti: la durata della sessione è determinata esclusivamente da quanto attivamente si comporta l'utente. È importante ricordare che tutti i tipi di finestre si basano su etichette di data/ora dei registri, non sull'ora di sistema.
Successivamente, realizzeremo la nostra topologia con ciascun tipo di finestra. Il codice completo verrà fornito solo nel primo esempio; per gli altri tipi di finestra, non ci sarà nulla da cambiare, tranne il tipo di operazione della finestra.
Finestre di sessione
Le finestre di sessione differiscono notevolmente da tutti gli altri tipi di finestre. Sono limitate non tanto dal tempo quanto dall'attività dell'utente (o dall'attività dell'entità che si desidera 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ù piccola si fonderà con la sessione a sinistra. Una 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 timestamp dai registri per determinare a quale sessione si riferisce la registrazione.
Utilizzo delle finestre di sessione per il monitoraggio delle transazioni di borsa
Utilizziamo le finestre di sessione per catturare informazioni sulle transazioni di borsa. L'implementazione delle finestre di sessione è mostrata nel listing 5.5 (che si trova nel file src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java).
La maggior parte delle operazioni di questa topologia le avete già incontrate, quindi non c'è bisogno di rivederle qui. Ma ci sono anche alcuni nuovi elementi che discuteremo ora.
In ogni operazione groupBy, di solito viene eseguita qualche operazione di aggregazione (aggregazione, riduzione o conteggio). Si può eseguire un'aggregazione cumulativa con un risultato progressivo, o un'aggregazione a finestra, in cui vengono considerati i registri all'interno di una finestra temporale specificata.
Il codice dell'elenco 5.5 conta il numero di 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. Un intervallo di inattività di 20 secondi significa che l'applicazione includerà qualsiasi registrazione che arriverà entro 20 secondi dalla fine o dall'inizio della sessione corrente nella sessione attiva.
Successivamente indichiamo quale operazione di aggregazione deve essere eseguita nella finestra di sessione — in questo caso count. Se la registrazione in ingresso esce oltre l'intervallo di inattività (da ciascun lato del timestamp), l'applicazione crea una nuova sessione. L'intervallo di conservazione significa mantenere la sessione per un certo periodo e consente dati tardivi che superano il periodo di inattività della sessione, ma che possono ancora essere uniti. Inoltre, l'inizio e la fine della nuova sessione, risultante dalla fusione, corrispondono al timestamp più presto e a quello più recente.
Esaminiamo alcune voci del metodo count per vedere come funzionano le sessioni (tab. 5.1).
Quando arrivano le voci, cerchiamo sessioni già esistenti con la stessa chiave, un orario di fine precedente all'attuale timestamp e un orario di inizio successivo all'attuale timestamp + intervallo di inattività. Tenendo conto di ciò, quattro voci della tab. 5.1 si uniscono in un'unica sessione nel modo seguente.
1. La prima voce è la 1, quindi l'orario di inizio è uguale all'orario di fine ed è 00:00:00.
2. Successivamente arriva la voce 2 e cerchiamo sessioni che finiscono non prima delle 23:59:55 e iniziano non oltre le 00:00:35. Troviamo la voce 1 e uniamo le sessioni 1 e 2. Prendiamo l'orario di inizio della sessione 1 (più precedente) e l'orario di fine della sessione 2 (più tardivo), così la nostra nuova sessione inizia alle 00:00:00 e termina alle 00:00:15.
3. Arriva la voce 3, cerchiamo sessioni tra le 00:00:30 e le 00:01:10 e non troviamo alcuna. Aggiungiamo la seconda sessione per la chiave 123-345-654,FFBE, che inizia e finisce alle 00:00:50.
4. Arriva la registrazione 4 e cerchiamo sessioni tra le 23:59:45 e le 00:00:25. Questa volta sono presenti entrambe le sessioni — 1 e 2. Tutte e tre le sessioni vengono unite in una sola, con l'orario di inizio alle 00:00:00 e l'orario di fine alle 00:00:15.
Dalla narrazione in questa sezione, è importante ricordare i seguenti dettagli:
- le sessioni non sono finestre di dimensioni fisse. La durata di una sessione è determinata dall'attività all'interno di un determinato intervallo temporale;
- i timestamp nei dati determinano se un evento rientra in una sessione esistente o in un intervallo di inattività.
Discuteremo ora il prossimo tipo di finestre — le finestre 'tumbling'.
Finestre 'tumbling'
Le finestre 'tumbling' catturano eventi che rientrano in un determinato intervallo di tempo. Immaginate di dover catturare tutte le transazioni di borsa di un'azienda ogni 20 secondi, quindi raccogliete tutti gli eventi in quel periodo. Al termine dell'intervallo di 20 secondi, la finestra 'tumbling' si sposta e passa a un nuovo intervallo di osservazione di 20 secondi. La figura 5.14 illustra questa situazione.
Come potete vedere, tutti gli eventi verificatisi negli ultimi 20 secondi sono inclusi nella finestra. Al termine di questo intervallo di tempo viene creata una nuova finestra.
Nel listing 5.6 è mostrato il codice che dimostra l'uso delle finestre 'cascading' per catturare ogni 20 secondi le transazioni di borsa (lo potete trovare 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 'cascading'. In questo esempio non c'è la chiamata al metodo until(), pertanto verrà utilizzato l'intervallo di salvataggio predefinito, pari a 24 ore.
Infine, è tempo di passare all'ultimo tipo di finestre — le finestre 'hopping'.
Finestre scorrevoli ('hopping')
Le finestre scorrevoli/'hopping' (sliding/hopping) sono simili a quelle 'cascading', 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 le finestre "sliding" e "tumbling", torniamo all'esempio del conteggio delle transazioni di borsa. Il nostro obiettivo è ancora contare il numero di transazioni, ma non vogliamo aspettare 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. Così avremo tre finestre di risultati con dati sovrapposti.
Nella listing 5.7 è fornito il codice per impostare finestre scorrevoli (che si può trovare nel file src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).
Una finestra "tumbling" può essere trasformata in una "sliding" mediante l'aggiunta della chiamata al metodo advanceBy(). Nell'esempio fornito, l'intervallo di conservazione è di 15 minuti.
In questa sezione hai visto come limitare i risultati dell'aggregazione con finestre temporali. In particolare, vorremmo che tu ricordassi tre cose da questa sezione:
- la dimensione delle finestre di sessione è limitata non dal lasso di tempo, ma dall'attività degli utenti;
- le finestre 'saltellanti' forniscono un'idea degli eventi all'interno di un determinato intervallo di tempo;
- la durata di funzionamento delle finestre 'saltellanti' è fissata, ma vengono aggiornate frequentemente e possono contenere in tutte le finestre registrazioni sovrapposte.
In seguito scopriremo come convertire KTable di nuovo in KStream per effettuare il join.
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 può essere necessario per la seguente semplice ragione. KStream è un flusso di registrazioni, mentre KTable è un flusso di aggiornamenti delle registrazioni, ma a volte potrebbe essere necessario aggiungere un contesto aggiuntivo al flusso di registrazioni utilizzando aggiornamenti provenienti da KTable.
Prendiamo i dati sul numero di transazioni di borsa e li uniamo alle notizie di borsa per le relative industrie. Ecco cosa occorre fare per raggiungere questo obiettivo tenendo conto del codice già esistente.
- Convert the KTable object with data on the number of stock transactions into a KStream, replacing the key with one that represents the industry corresponding to the stock symbol.
- Create a KTable object that reads data from the stock news topic. This new KTable will be categorized by industries.
- Combine news updates with information about the number of stock transactions by industry.
Now let's see how to implement this action plan.
Transforming KTable into KStream
To convert KTable into KStream, you need to do the following.
- Invoke the KTable.toStream() method.
- Use the KStream.map method to replace the key with the name of the industry, and then extract the TransactionSummary object from the Windowed instance.
We will chain these operations as follows (the code can be found in the file src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (listing 5.8).
Since we are performing the KStream.map operation, the re-partitioning for the returned KStream instance is done automatically when it is used in the join.
Abbiamo completato il processo di conversione; 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 si trova nel file src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (listing 5.9).
È importante notare che non è necessario specificare alcun oggetto Serde, poiché nelle impostazioni vengono utilizzati Serde di tipo stringa. Inoltre, grazie all'uso dell'enumerazione EARLIEST, la tabella viene riempita con i registri fin dall'inizio.
Ora possiamo passare all'ultimo passaggio: la connessione.
Connessione degli aggiornamenti delle notizie con i dati sul numero di transazioni
Creare la connessione non è complicato. Utilizzeremo una connessione sinistra nel caso in cui non ci siano notizie di borsa per l'industria corrispondente (il codice necessario si trova nel file src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (listing 5.10).
Questo operatore leftJoin è abbastanza semplice. A differenza delle join del capitolo 4, il metodo JoinWindow non viene utilizzato, poiché nell'esecuzione della join KStream-KTable per ogni chiave in KTable è presente solo una registrazione. Questo tipo di join non è limitato nel tempo: la registrazione è o presente in KTable o assente. La conclusione principale è che gli oggetti KTable possono arricchire KStream con dati di riferimento meno aggiornati.
Ora considereremo un modo più efficace per arricchire gli eventi da KStream.
5.3.4. Oggetti GlobalKTable
Come avete capito, c'è la necessità di arricchire i flussi di eventi o di aggiungere contesto ad essi. Nel capitolo 4 avete visto connessioni tra due oggetti KStream, e nella sezione precedente - la connessione tra KStream e KTable. In tutti questi casi è necessario un nuovo partizionamento del flusso di dati per mappare le chiavi a un nuovo tipo o valore. A volte il partizionamento è eseguito esplicitamente, altre volte Kafka Streams lo porta a termine automaticamente. Il partizionamento è necessario poiché le chiavi sono cambiate e i record devono trovarsi in nuove partizioni; altrimenti, la connessione non sarà possibile (discusso nel capitolo 4, nella sezione “Partizionamento dei dati” sotto-sezione 4.2.4).
Il partizionamento ha il suo costo
La riconfigurazione richiede costi aggiuntivi — costi di risorse per la creazione di argomenti intermedi, mantenendo 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 il join su più di un aspetto o dimensione, è necessario organizzare i join in catena, visualizzare le registrazioni con nuove chiavi e ripetere il processo di riconfigurazione.
Join con set di dati di dimensioni inferiori
In alcuni casi, il volume dei dati di riferimento con cui si prevede di eseguire il join è relativamente piccolo, quindi le loro copie complete possono effettivamente essere memorizzate localmente su ciascun nodo. Per tali situazioni, Kafka Streams prevede una classe GlobalKTable.
Le istanze di GlobalKTable sono uniche, poiché l'applicazione replica tutti i dati su ciascun nodo. E poiché tutti i dati sono presenti su ciascun nodo, non è necessario partizionare il flusso di eventi in base alla chiave dei dati di riferimento, rendendolo accessibile a tutte le sezioni. Inoltre, con gli oggetti GlobalKTable, è possibile effettuare connessioni senza chiave. Torniamo a uno degli esempi precedenti per dimostrare questa possibilità.
Collegamento di oggetti KStream con oggetti GlobalKTable
Nel sottosezione 5.3.2 abbiamo effettuato un'aggregazione a finestra delle transazioni di borsa per i clienti. I risultati di questa aggregazione apparivano circa come segue:
{customerId='074-09-3705', stockTicker='GUTM'}, 17
{customerId='037-34-5184', stockTicker='CORK'}, 16Sebbene questi risultati soddisfacessero l'obiettivo, sarebbe stato più conveniente se venissero visualizzati anche il nome del cliente e la denominazione completa dell'azienda. Per aggiungere il nome del cliente e il nome dell'azienda, è possibile effettuare normali collegamenti, ma sarà necessario eseguire due mappature delle chiavi e un'ulteriore partizione. Con GlobalKTable, è possibile evitare i costi associati a tali operazioni.
Per questo utilizzeremo l'oggetto countStream dell'elenco 5.11 (il codice corrispondente può essere trovato nel file src/main/java/bbejeck/chapter_5/GlobalKTableExample.java), collegandolo a due oggetti GlobalKTable.
Ne abbiamo già parlato in precedenza, quindi non mi ripeterò. Ma voglio sottolineare che il codice nella funzione toStream().map è astratto in un oggetto funzione per migliorare la leggibilità invece di utilizzare un'espressione lambda inline.
Il passo successivo è dichiarare due istanze di GlobalKTable (il codice fornito può essere trovato nel file src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (elenco 5.12).
Si noti che i nomi dei topic sono descritti utilizzando tipi enumerativi.
Ora che abbiamo preparato tutti i componenti, resta solo da scrivere il codice per il join (che può essere trovato nel file src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (elenco 5.13).
Anche se in questo codice sono presenti due join, questi sono organizzati in una catena, poiché nessun risultato singolo viene utilizzato separatamente. I risultati vengono restituiti alla fine dell'intera operazione.
Eseguendo l'operazione di join sopra indicata, otterrete risultati del tipo:
{customer='Barney, Smith' company="Exxon", transactions= 17}La sostanza non è cambiata, ma questi risultati sono più chiari.
Se consideriamo il capitolo 4, hai già visto diversi tipi di connessioni in azione. Sono elencati nella tabella 5.2. Questa tabella riflette le capacità di connessione valide per la versione 1.0.0 di Kafka Streams; in futuri rilasci potrebbero esserci delle modifiche.
In conclusione, ricordo l'essenziale: puoi unire flussi di eventi (KStream) e flussi di aggiornamenti (KTable) utilizzando uno stato locale. Inoltre, se la dimensione dei dati di riferimento non è troppo grande, puoi utilizzare l'oggetto GlobalKTable. GlobalKTable replica tutte le partizioni su ciascun nodo dell'applicazione Kafka Streams, garantendo così la disponibilità di tutti i dati indipendentemente dalla partizione a cui appartiene la chiave.
In seguito vedremo una funzionalità di Kafka Streams che consente di monitorare le modifiche di stato senza consumare dati dal topic Kafka.
5.3.5. Stato disponibile per le query
Abbiamo già eseguito diverse operazioni con 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 registriamo i risultati in un topic, è necessario utilizzare un consumatore Kafka per visualizzarli.
La lettura dei dati da questi topic può essere considerata una forma di viste materializzate. Per i nostri scopi, possiamo utilizzare la definizione di vista materializzata da «Wikipedia»: «…un oggetto fisico del database che contiene i risultati dell'esecuzione di una query. 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 pivot ottenuta tramite aggregazione» (https://en.wikipedia.org/wiki/Materialized_view).
Kafka Streams consente anche di eseguire query interattive sui negozi di stato, offrendo la possibilità di leggere direttamente queste rappresentazioni materializzate. È importante notare che la query al negozio di stato è un'operazione "solo lettura". Questo significa che non rischiate di rendere lo stato incoerente durante l'elaborazione dei dati da parte dell'applicazione.
La possibilità di effettuare query dirette sui negozi di stato è fondamentale. Significa che è possibile creare applicazioni — dashboard — senza dover prima recuperare i dati dal consumatore Kafka. Questo aumento di efficienza dell'applicazione è possibile poiché non è necessario riscrivere i dati:
- grazie alla località dei dati, è possibile accedervi rapidamente;
- si elimina la duplicazione dei dati, poiché non vengono registrati in uno storage esterno.
La cosa principale che vorrei che tu ricordassi è: puoi eseguire direttamente query sugli stati dall'applicazione. Non si può sottovalutare le opportunità che questo ti offre. Invece di consumare dati da Kafka e archiviare registrazioni nel database per l'applicazione, puoi eseguire query sugli archivi di stato con lo stesso risultato. Le richieste dirette agli archivi di stato significano meno codice (assenza di un consumatore) e meno software (non è necessario avere una tabella nel database per archiviare i risultati).
Abbiamo coperto una notevole quantità di informazioni in questo capitolo, quindi per il momento interromperemo la nostra discussione sulle query interattive sugli archivi di stato. Ma non preoccuparti: nel capitolo 9 costruiremo una semplice applicazione — un cruscotto con query interattive. Per dimostrare le query interattive e le possibilità di integrarle nelle applicazioni Kafka Streams, utilizzeremo alcuni degli esempi di questo capitolo e di quelli precedenti.
Riepilogo
- Gli oggetti KStream rappresentano flussi di eventi, simili a inserimenti in un database. Gli oggetti KTable rappresentano flussi di aggiornamenti, sono più simili ad aggiornamenti nel database. La dimensione dell'oggetto KTable non aumenta, le vecchie registrazioni vengono sostituite da nuove.
- Gli oggetti KTable sono necessari per le operazioni di aggregazione.
- Con le operazioni finestra è possibile suddividere i dati aggregati in ceste temporali.
- Grazie agli oggetti GlobalKTable è possibile accedere ai dati di riferimento da qualsiasi punto dell'applicazione, indipendentemente dalla suddivisione in partizioni.
- Sono possibili connessioni tra oggetti KStream, KTable e GlobalKTable.
Fino ad ora ci siamo concentrati sulla creazione di applicazioni Kafka Streams utilizzando l'DSL KStream di alto livello. Anche se l'approccio ad alto livello consente di scrivere programmi puliti e concisi, il suo utilizzo comporta un certo compromesso. Lavorare con l'DSL KStream significa aumentare la sinteticità del codice a scapito del controllo. Nel capitolo successivo esploreremo l'API di basso livello dei nodi elaboratori e considereremo altri compromessi. I programmi diventeranno più lunghi rispetto a prima, ma avremo la possibilità di creare praticamente qualsiasi nodo elaboratore di cui potremmo avere bisogno.
→ Maggiori dettagli sulla disponibilità del libro si possono trovare su
→ Per gli utenti di Habr, sconto del 25% con il coupon — Kafka Streams
→ Dopo il pagamento della versione cartacea del libro, verrà inviata una copia elettronica via email.
Fonte: habr.com
