{"id":35786,"date":"2019-10-31T22:06:19","date_gmt":"2019-10-31T19:06:19","guid":{"rendered":"https:\/\/prohoster.info\/blog\/kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni\/"},"modified":"2019-10-31T22:06:19","modified_gmt":"2019-10-31T19:06:19","slug":"kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni","status":"publish","type":"post","link":"https:\/\/prohoster.info\/it\/blog\/administrirovanie\/kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni","title":{"rendered":"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"<p><noindex><a rel=\"nofollow\" href=\"https:\/\/habr.com\/ru\/company\/piter\/blog\/457756\/\"><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/6f8bd2b31b87b0c760c1148515893c43.jpg\" style=\"display:block;margin: 0 auto;\" \/><\/a><\/noindex> Ciao, Habrozhiteli! Questo libro \u00e8 adatto a qualsiasi sviluppatore che desidera comprendere l'elaborazione dei flussi. Comprendere la programmazione distribuita aiuter\u00e0 a studiare meglio Kafka e Kafka Streams. Sarebbe utile conoscere anche il framework Kafka stesso, ma non \u00e8 obbligatorio: ti spiegher\u00f2 tutto ci\u00f2 che \u00e8 necessario. Gli sviluppatori esperti di Kafka, cos\u00ec 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\u00e0 abituati a concetti come la serializzazione, impareranno a applicare le proprie competenze per creare applicazioni Kafka Streams. Il codice sorgente del libro \u00e8 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\u00e0 utile.<br \/>\n<noindex><a rel=\"nofollow\" name=\"habracut\"><\/a><\/noindex><\/p>\n<h3>Estratto. 5.3. Aggregazione e operazioni a finestre<\/h3>\n<p>\nIn questa sezione ci concentreremo sullo studio delle parti pi\u00f9 promettenti di Kafka Streams. Fino ad ora abbiamo esaminato i seguenti aspetti di Kafka Streams:<\/p>\n<ul>\n<li>creazione di una topologia di elaborazione;<\/li>\n<li>uso dello stato nelle applicazioni di flusso;<\/li>\n<li>esecuzione di unioni di flussi di dati;<\/li>\n<li>differenze tra flussi di eventi (KStream) e flussi di aggiornamenti (KTable).<\/li>\n<\/ul>\n<p>\nNei prossimi esempi uniremo tutti questi elementi. Inoltre, ti familiarizzerai con le operazioni a finestre, un'altra fantastica possibilit\u00e0 delle applicazioni di flusso. Il nostro primo esempio sar\u00e0 una semplice aggregazione.<\/p>\n<h3>5.3.1. Aggregazione del volume delle vendite di azioni per settori industriali<\/h3>\n<p>\nL'aggregazione e il raggruppamento sono strumenti essenziali quando si lavora con dati di flusso. Esaminare singoli record man mano che arrivano spesso non \u00e8 sufficiente. Per estrarre ulteriori informazioni dai dati, \u00e8 necessario raggrupparli e combinarli.<\/p>\n<p>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.<\/p>\n<p>Per tale aggregazione saranno necessari diversi passaggi per trasformare i dati nella forma desiderata (parlando in termini generali).<\/p>\n<ol>\n<li>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 \u00e8 che l'oggetto StockTransaction contiene metadati sulle vendite, mentre a noi servono solo i dati sul numero di azioni vendute.<\/li>\n<li>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, \u00e8 necessario chiamare successivamente il metodo KGroupedStream.reduce.<\/li>\n<\/ol>\n<p><\/p>\n<blockquote><p><b>Che cos'\u00e8 l'interfaccia KGroupedStream<\/b><\/p>\n<p>I metodi KStream.groupBy e KStream.groupByKey restituiscono un'istanza di KGroupedStream. KGroupedStream \u00e8 una rappresentazione intermedia di un flusso di eventi dopo il raggruppamento per chiavi. Non \u00e8 assolutamente destinato a un utilizzo diretto. Invece, KGroupedStream \u00e8 utilizzato per operazioni di aggregazione, il cui risultato \u00e8 sempre un KTable. Poich\u00e9 il risultato delle operazioni di aggregazione \u00e8 un KTable e in esse viene applicato uno store di stato, potrebbe non tutti gli aggiornamenti risultanti vengano inviati ulteriormente lungo la pipeline.<\/p>\n<p>Il metodo KTable.groupBy restituisce un KGroupedTable simile: una rappresentazione intermedia di un flusso di aggiornamenti riorganizzati per chiave.<\/p><\/blockquote>\n<p>\nFermiamoci un attimo e diamo un'occhiata alla figura 5.9, che mostra cosa abbiamo ottenuto. Questa topologia dovrebbe esserti gi\u00e0 ben nota.<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/9fd61317cde376362adcaeec72908919.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nDiamo ora un'occhiata al codice per questa topologia (che puoi trovare nel file src\/main\/java\/bbejeck\/chapter_5\/AggregationsAndReducingExample.java) (listing 5.2).<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/f937287e448295fbd467c283ceca316a.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nIl codice fornito \u00e8 conciso e compie una grande quantit\u00e0 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'\u00e8 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.<\/p>\n<blockquote><p><b>GroupByKey e GroupBy<\/b><\/p>\n<p>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 \u00e8 la differenza tra di loro e quando usare ciascuno?<\/p>\n<p>Il metodo GroupByKey viene utilizzato quando le chiavi in KStream non sono vuote. E soprattutto, il flag \"richiede una nuova partizione\" non \u00e8 mai stato impostato.<\/p>\n<p>Il metodo GroupBy presuppone che tu abbia cambiato le chiavi per la raggruppamento, quindi il flag di ri-partizionamento \u00e8 impostato su true. Eseguire connessioni, aggregazioni, ecc. dopo il metodo GroupBy porter\u00e0 a una nuova partizione automatica.<br \/>\nIn sintesi: si dovrebbe utilizzare GroupByKey ogni volta che \u00e8 possibile, piuttosto che GroupBy.<\/p><\/blockquote>\n<p>\nCosa fanno i metodi mapValues e groupBy \u00e8 chiaro, quindi diamo un'occhiata al metodo sum() (puoi trovarlo nel file src\/main\/java\/bbejeck\/model\/ShareVolume.java) (listing 5.3).<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/8e9a6f873594f9b5fef9a96c42353a61.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nIl 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. \u00c8 importante non dimenticare che tutti gli aggiornamenti si riflettono nel precedente shareVolumeKTable, ma non tutti vengono inviati ulteriormente.<\/p>\n<p>Successivamente, utilizziamo questo KTable per eseguire un'aggregazione (in base alla quantit\u00e0 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.<\/p>\n<ol>\n<li>Eseguire un'altra operazione groupBy per raggruppare i singoli oggetti ShareVolume per settori industriali.<\/li>\n<li>Iniziare a sommare gli oggetti ShareVolume. Questa volta l'oggetto di aggregazione rappresenta una coda di priorit\u00e0 di dimensioni fisse. In questa coda di dimensioni fisse vengono mantenute solo cinque aziende con le maggiori quantit\u00e0 di azioni vendute.<\/li>\n<li>Mappare le code del punto precedente in un valore stringa e restituire le cinque pi\u00f9 vendute per numero di azioni per settori industriali.<\/li>\n<li>Scrivere i risultati in formato stringa nel topic.<\/li>\n<\/ol>\n<p>\nLa figura 5.10 mostra il grafo della topologia del flusso dei dati. Come puoi vedere, il secondo ciclo di elaborazione \u00e8 piuttosto semplice.<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/dabd1507eee267038edb7f8d76d8d8ae.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nOra, 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).<\/p>\n<p>In questo inizializzatore c'\u00e8 una variabile fixedQueue. \u00c8 un oggetto personalizzato - un adattatore per java.util.TreeSet, utilizzato per tenere traccia dei N migliori risultati in ordine decrescente della quantit\u00e0 di azioni vendute.<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/488b072b0d91b81c925ca72291e69e48.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nHai gi\u00e0 incontrato le chiamate groupBy e mapValues, quindi non ci soffermeremo su di esse (chiamiamo il metodo KTable.toStream, poich\u00e9 il metodo KTable.print \u00e8 considerato obsoleto). Ma non hai ancora visto la versione KTable del metodo aggregate(), quindi dedicheremo un po' di tempo a discuterne.<\/p>\n<p>Come ricordi, KTable si distingue perch\u00e9 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\u00e0 un'altra registrazione con la stessa chiave, la vecchia registrazione viene rimossa tramite un sottrattore (il terzo parametro nella chiamata al metodo aggregate).<\/p>\n<p>Questo significa che il nostro aggregatore, FixedSizePriorityQueue, non aggrega affatto tutti i valori con una chiave, ma memorizza la somma mobile delle quantit\u00e0 dei N tipi di azioni pi\u00f9 venduti. In ogni registrazione in arrivo \u00e8 contenuto il numero totale di azioni vendute fino a quel momento. KTable ti dar\u00e0 informazioni su quali aziende stanno vendendo il maggior numero di azioni in questo momento, non \u00e8 richiesta l'aggregazione mobile di ciascun aggiornamento.<\/p>\n<p>Abbiamo imparato a fare due cose importanti:<\/p>\n<ul>\n<li>raggruppare i valori in KTable in base a una chiave comune;<\/li>\n<li>eseguire su questi valori raggruppati operazioni utili come la compressione e l'aggregazione.<\/li>\n<\/ul>\n<p>\nEssere in grado di eseguire queste operazioni \u00e8 importante per comprendere il significato dei dati che scorrono attraverso l'applicazione Kafka Streams e scoprire quali informazioni essi portano.<\/p>\n<p>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\u00e9 \u00e8 cos\u00ec cruciale averlo: consente di tenere traccia delle informazioni che hai gi\u00e0 visualizzato. L'accesso locale permette di evitare ritardi di rete, rendendo cos\u00ec l'applicazione pi\u00f9 performante e resistente agli errori.<\/p>\n<p>Quando si esegue un'operazione di aggregazione o di riduzione, \u00e8 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 \u00e8 importante, poich\u00e9 le operazioni di aggregazione sono destinate a ottenere informazioni finali. Se non si utilizza lo stato locale, KTable invier\u00e0 avanti tutti i risultati di aggregazione e riduzione.<\/p>\n<p>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).<\/p>\n<h3>5.3.2. Operazioni a finestra<\/h3>\n<p>\nNella 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\u00f9 vendute sul mercato.<\/p>\n<p>A volte sono necessarie operazioni continue di aggregazione e riduzione dei risultati. Altre volte \u00e8 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\u00f2 eseguire tali operazioni ripetutamente, ma con risultati che riguardano solo gli intervalli di tempo specificati (finestre temporali).<\/p>\n<h3>Conteggio delle transazioni di borsa per cliente<\/h3>\n<p>\nNell'esempio seguente ci concentreremo sul monitoraggio delle transazioni di borsa per vari trader \u2014 sia grandi organizzazioni che astuti investitori solitari.<\/p>\n<p>Ci sono due possibili motivi per un simile monitoraggio. Uno di questi \u00e8 la necessit\u00e0 di sapere cosa acquistano\/vendono i leader di mercato. Se questi grandi attori e investitori esperti vedono opportunit\u00e0 emergere, ha senso seguire le loro strategie. Il secondo motivo \u00e8 il desiderio di notare eventuali segni di transazioni illegali che utilizzano informazioni riservate. Per questo, sar\u00e0 necessario analizzare la correlazione tra grandi picchi nelle vendite e importanti comunicati stampa.<\/p>\n<p>Questo monitoraggio consiste in fasi come:<\/p>\n<ul>\n<li>creazione di un flusso di lettura dal topic stock-transactions;<\/li>\n<li>raggruppamento delle registrazioni in arrivo per identificativo dell'acquirente e simbolo di borsa. La chiamata al metodo groupBy restituisce un'istanza della classe KGroupedStream;<\/li>\n<li>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;<\/li>\n<li>conteggio delle transazioni per l'operazione di aggregazione. Il flusso di dati a finestre determina se una specifica registrazione \u00e8 considerata in questo conteggio;<\/li>\n<li>registrazione dei risultati nel topic o stampa in console durante lo sviluppo.<\/li>\n<\/ul>\n<p>\nLa topologia di questa applicazione \u00e8 semplice, ma non guasta avere un'immagine chiara di essa. Diamo un'occhiata alla figura 5.11.<\/p>\n<p>Successivamente esamineremo la funzionalit\u00e0 delle operazioni su finestre e il codice corrispondente.<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/67d9d8d8acb1370a7c7fe5cd9a8b6aa7.jpg\" style=\"display:block;margin: 0 auto;\" \/><\/p>\n<h3>Tipi di finestre<\/h3>\n<p>\nIn Kafka Streams esistono tre tipi di finestre:<\/p>\n<ul>\n<li>finestre di sessione;<\/li>\n<li>finestre 'tumbling';<\/li>\n<li>finestre 'sliding'\/hopping.<\/li>\n<\/ul>\n<p>\nQuale 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 \u00e8 determinata esclusivamente dall'attivit\u00e0 dell'utente. L'importante \u00e8 non dimenticare che tutti i tipi di finestre si basano sui timestamp delle registrazioni, e non sull'orario di sistema.<\/p>\n<p>Successivamente implementeremo la nostra topologia con ciascuno dei tipi di finestre. Il codice completo verr\u00e0 fornito solo nel primo esempio; per gli altri tipi di finestre nulla cambier\u00e0, tranne il tipo dell'operazione di finestra.<\/p>\n<h3>Finestre di sessione<\/h3>\n<p>\nLe finestre di sessione si differenziano notevolmente da tutti gli altri tipi di finestre. Non sono limitate tanto nel tempo, quanto nell'attivit\u00e0 dell'utente (o nell'attivit\u00e0 di quell'entit\u00e0 che desideri monitorare). Le finestre di sessione sono delimitate da periodi di inattivit\u00e0.<\/p>\n<p>La figura 5.12 illustra il concetto di finestre di sessione. Una sessione pi\u00f9 breve si fonder\u00e0 con la sessione a sinistra. E la sessione a destra sar\u00e0 separata, in quanto segue un lungo periodo di inattivit\u00e0. Le finestre di sessione si basano sulle azioni degli utenti, ma utilizzano i timestamp dalle registrazioni per determinare a quale sessione riguarda la registrazione.<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/d39a5db7af7d6aa2194802622b2b47fd.jpg\" style=\"display:block;margin: 0 auto;\" \/><\/p>\n<h3>Utilizzo delle finestre di sessione per monitorare le transazioni di borsa<\/h3>\n<p>\nUtilizzeremo le finestre di sessione per catturare informazioni sulle transazioni di borsa. L'implementazione delle finestre di sessione \u00e8 mostrata nel listing 5.5 (che pu\u00f2 essere trovato nel file src\/main\/java\/bbejeck\/chapter_5\/CountingWindowingAndKTableJoinExample.java).<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/2dcbd9a36baec0e746aad165121451b3.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nLa maggior parte delle operazioni di questa topologia le hai gi\u00e0 incontrate, quindi non c'\u00e8 bisogno di rivederle qui. Ma ci sono anche alcuni nuovi elementi che discuteremo ora.<\/p>\n<p>In ogni operazione groupBy viene solitamente eseguita qualche operazione di aggregazione (aggregazione, riduzione o conteggio). \u00c8 possibile eseguire un'aggregazione cumulativa in modo incrementale, oppure un'aggregazione finestrata, che considera le registrazioni all'interno di una finestra temporale specifica.<\/p>\n<p>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.<\/p>\n<p>Con la chiamata windowedBy(SessionWindows.with(twentySeconds).until(fifteenMinutes)) creiamo una finestra di sessione con un intervallo di inattivit\u00e0 di 20 secondi e un intervallo di conservazione di 15 minuti. L'intervallo di inattivit\u00e0 di 20 secondi significa che l'applicazione includer\u00e0 qualsiasi registrazione che arriva entro 20 secondi dalla fine o dall'inizio della sessione corrente nella sessione attuale (attiva).<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/9bd47b04698086872fd135b4c67eb938.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nSuccessivamente, 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\u00e0 (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\u00e0 della sessione, ma possono ancora essere collegati. Inoltre, l'inizio e la fine della nuova sessione risultante dalla fusione corrispondono al timestamp pi\u00f9 antico e a quello pi\u00f9 recente.<\/p>\n<p>Esaminiamo alcuni record dal metodo count per vedere come funzionano le sessioni (tab. 5.1).<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/ec04aae466d88c2d2349474069c8d541.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nAll'arrivo dei record, cerchiamo sessioni gi\u00e0 esistenti con la stessa chiave, con un orario di fine inferiore all'attuale timestamp \u2014 intervallo di inattivit\u00e0 \u2014 e un orario di inizio superiore all'attuale timestamp + intervallo di inattivit\u00e0. Tenendo conto di questo, quattro record della tab. 5.1 si fondono in un'unica sessione nel seguente modo.<\/p>\n<p>1. Per prima cosa arriva il record 1, quindi l'orario di inizio \u00e8 uguale a quello di fine ed \u00e8 00:00:00.<\/p>\n<p>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\u00f9 antico) e l'orario di fine della sessione 2 (pi\u00f9 recente), cos\u00ec la nostra nuova sessione inizia alle 00:00:00 e termina alle 00:00:15.<\/p>\n<p>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.<\/p>\n<p>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.<\/p>\n<p>Dalle informazioni presentate in questa sezione, \u00e8 importante ricordare i seguenti punti chiave:<\/p>\n<ul>\n<li>le sessioni non sono finestre di dimensioni fisse. La durata della sessione \u00e8 determinata dall'attivit\u00e0 nell'ambito di un determinato intervallo di tempo;<\/li>\n<li>i timestamp nei dati determinano se un evento rientra in una sessione esistente o in un periodo di inattivit\u00e0.<\/li>\n<\/ul>\n<p>\nSuccessivamente discuteremo il prossimo tipo di finestre: le finestre 'capovolte'.<\/p>\n<h3>'Finestre capovolte'<\/h3>\n<p>\nLe finestre \u00abtumbling\u00bb 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 \u00abtumbling\u00bb si rovescia e passa a un nuovo intervallo di 20 secondi di osservazione. La figura 5.14 illustra questa situazione.<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/e90e560d9ddda5e2e524b7c387ad9874.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nCome 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.<\/p>\n<p>Nel listing 5.6 \u00e8 fornito il codice che dimostra l'uso delle finestre \u00abtumbling\u00bb per catturare ogni 20 secondi le transazioni di borsa (puoi trovarlo nel file src\/main\/java\/bbejeck\/chapter_5\/CountingWindowingAndKtableJoinExample.java).<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/e055bb1b288c7d500b64372fe3fbf064.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nGrazie a questa piccola modifica alla chiamata del metodo TimeWindows.of, \u00e8 possibile utilizzare una finestra \u00abtumbling\u00bb. In questo esempio non c'\u00e8 chiamata al metodo until(), quindi verr\u00e0 utilizzato l'intervallo di conservazione predefinito, che \u00e8 di 24 ore.<\/p>\n<p>Infine, \u00e8 ora di passare all'ultima delle opzioni di finestra: le finestre \u00abhopping\u00bb.<\/p>\n<h3>Finestre scorrevoli (\u00abhopping\u00bb)<\/h3>\n<p>\nLe finestre scorrevoli\/\u00abhopping\u00bb sono simili alle finestre \u00abtumbling\u00bb, 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.<\/p>\n<p>Per illustrare le differenze tra finestre \u00abtumbling\u00bb e \u00abhopping\u00bb, 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\u00f9 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.<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/415c8cd9f2b60d453a1a01c3bc99331f.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nNel listing 5.7 \u00e8 fornito il codice per definire finestre scorrevoli (puoi trovarlo nel file src\/main\/java\/bbejeck\/chapter_5\/CountingWindowingAndKtableJoinExample.java).<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/ab2d1a64380d256d3fb084e16597417c.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nLa finestra \"rotolante\" pu\u00f2 essere trasformata in una finestra \"rimbalzante\" aggiungendo una chiamata al metodo advanceBy(). Nell'esempio fornito, l'intervallo di salvataggio \u00e8 di 15 minuti.<\/p>\n<p>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:<\/p>\n<ul>\n<li>la dimensione delle finestre di sessione \u00e8 limitata non da un intervallo di tempo, ma dall'attivit\u00e0 degli utenti;<\/li>\n<li>le finestre \"rotolanti\" offrono una panoramica degli eventi all'interno di un determinato periodo di tempo;<\/li>\n<li>la durata delle finestre \"rimbalzanti\" \u00e8 fissa, ma vengono frequentemente aggiornate e possono contenere record sovrapposti in tutte le finestre.<\/li>\n<\/ul>\n<p>\nSuccessivamente scopriremo come riconvertire un KTable in KStream per l'unione.<\/p>\n<h3>5.3.3. Unione di oggetti KStream e KTable<\/h3>\n<p>\nNel 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 \u00e8 un flusso di record, mentre KTable \u00e8 un flusso di aggiornamenti, ma a volte pu\u00f2 essere necessario aggiungere un contesto supplementare al flusso di record tramite aggiornamenti da KTable.<\/p>\n<p>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\u00e0 esistente.<\/p>\n<ol>\n<li>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.<\/li>\n<li>Creare un oggetto KTable che legga i dati da un topic con le notizie di borsa. Questo nuovo KTable sar\u00e0 categorizzato per settori industriali.<\/li>\n<li>Unire gli aggiornamenti delle notizie con le informazioni sul numero di transazioni di borsa per i settori industriali.<\/li>\n<\/ol>\n<p>\nOra vediamo come implementare questo piano d'azione.<\/p>\n<h3>Trasformazione di KTable in KStream<\/h3>\n<p>\nPer trasformare KTable in KStream, \u00e8 necessario fare quanto segue.<\/p>\n<ol>\n<li>Chiamare il metodo KTable.toStream().<\/li>\n<li>Utilizzando la chiamata al metodo KStream.map, sostituire la chiave con il nome del settore industriale, dopodich\u00e9 estrarre l'oggetto TransactionSummary dall'istanza Windowed.<\/li>\n<\/ol>\n<p>\nCollegheremo queste operazioni in una catena nel seguente modo (il codice pu\u00f2 essere trovato nel file src\/main\/java\/bbejeck\/chapter_5\/CountingWindowingAndKtableJoinExample.java) (elenco 5.8).<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/0d43c2650f6e66e2816ed383da3a29c2.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nPoich\u00e9 eseguiamo l'operazione KStream.map, il ri-partizionamento per l'istanza KStream restituita viene effettuato automaticamente durante il suo utilizzo in una join.<\/p>\n<p>Abbiamo completato il processo di trasformazione, ora dobbiamo creare un oggetto KTable per leggere le notizie di borsa.<\/p>\n<h3>Creazione di KTable per le notizie di borsa<\/h3>\n<p>\nFortunatamente, per creare un oggetto KTable \u00e8 sufficiente una sola riga di codice (questo codice pu\u00f2 essere trovato nel file src\/main\/java\/bbejeck\/chapter_5\/CountingWindowingAndKtableJoinExample.java) (listing 5.9).<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/6e83a393fdc9ab74fda4cbdddddb5213.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\n\u00c8 importante notare che non \u00e8 necessario specificare oggetti Serde, poich\u00e9 nelle impostazioni vengono utilizzati Serde per stringhe. Inoltre, grazie all'uso dell'enumerazione EARLIEST, la tabella viene popolata con le registrazioni fin dall'inizio.<\/p>\n<p>Ora possiamo passare all'ultimo passo: la join.<\/p>\n<h3>Join tra aggiornamenti delle notizie e dati sul numero di transazioni<\/h3>\n<p>\nCreare una join non presenta difficolt\u00e0. Utilizzeremo una join sinistra nel caso in cui non ci siano notizie di borsa per l'industria corrispondente (il codice necessario pu\u00f2 essere trovato nel file src\/main\/java\/bbejeck\/chapter_5\/CountingWindowingAndKtableJoinExample.java) (listing 5.10).<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/92ed70f98927d2f778ad14dbc2a5aa26.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nQuesto operatore leftJoin \u00e8 piuttosto semplice. A differenza delle join del Capitolo 4, il metodo JoinWindow non viene utilizzato, poich\u00e9 durante l'esecuzione della join KStream-KTable esiste solo una registrazione per ogni chiave in KTable. Questa join non \u00e8 limitata nel tempo: la registrazione \u00e8 presente in KTable o non lo \u00e8 affatto. La conclusione principale: mediante gli oggetti KTable \u00e8 possibile arricchire KStream con dati di riferimento meno frequentemente aggiornati.<\/p>\n<p>Ora esamineremo un metodo pi\u00f9 efficiente per arricchire eventi da KStream.<\/p>\n<h3>5.3.4. Oggetti GlobalKTable<\/h3>\n<p>\nCome avete capito, c'\u00e8 la necessit\u00e0 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 \u2013 la join tra KStream e KTable. In tutti questi casi, \u00e8 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 \u00e8 necessario poich\u00e9 le chiavi sono cambiate e le registrazioni devono trovarsi nelle nuove partizioni, altrimenti la join non sar\u00e0 possibile (questo \u00e8 stato discusso nel capitolo 4, nella sezione \"Ri-partizionamento dei dati\" del sotto-capitolo 4.2.4).<\/p>\n<h3>La ricostituzione dei segmenti ha un costo<\/h3>\n<p>\nLa ricostituzione dei segmenti richiede risorse \u2014 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 \u00e8 necessario eseguire un'unione su pi\u00f9 di un aspetto o dimensione, \u00e8 necessario organizzare le unioni in serie, mappare le voci con nuove chiavi e ripetere il processo di ricostituzione dei segmenti.<\/p>\n<h3>Unione con set di dati di dimensioni minori<\/h3>\n<p>\nIn alcuni casi, il volume dei dati di riferimento con cui si prevede di unire \u00e8 relativamente ridotto, quindi le loro copie complete possono essere memorizzate localmente su ciascun nodo. Per tali situazioni, Kafka Streams prevede la classe GlobalKTable.<\/p>\n<p>Le istanze di GlobalKTable sono uniche, poich\u00e9 l'applicazione replica tutti i dati su ciascun nodo. Poich\u00e9 su ciascun nodo sono presenti tutti i dati, non \u00e8 necessario segmentare il flusso di eventi in base alla chiave dei dati di riferimento affinch\u00e9 sia accessibile a tutte le sezioni. Con gli oggetti GlobalKTable \u00e8 possibile eseguire anche unioni senza chiave. Torniamo a uno dei precedenti esempi per dimostrare questa capacit\u00e0.<\/p>\n<h3>Unione di oggetti KStream con oggetti GlobalKTable<\/h3>\n<p>\nNella sezione 5.3.2 abbiamo eseguito aggregazioni finestrate delle transazioni di borsa per clienti. I risultati di questa aggregazione erano simili ai seguenti:<\/p>\n<pre><code class=\"plaintext\">{customerId='074-09-3705', stockTicker='GUTM'}, 17\n{customerId='037-34-5184', stockTicker='CORK'}, 16<\/code><\/pre>\n<p>\nSebbene questi risultati soddisfacessero l'obiettivo fissato, sarebbe stato pi\u00f9 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, \u00e8 possibile eseguire normali unioni, ma sar\u00e0 necessario effettuare due mappature delle chiavi e ricostituire i segmenti. Con GlobalKTable \u00e8 possibile evitare costi per tali operazioni.<\/p>\n<p>A tal fine, utilizzeremo l'oggetto countStream dal listing 5.11 (il codice corrispondente pu\u00f2 essere trovato nel file src\/main\/java\/bbejeck\/chapter_5\/GlobalKTableExample.java), unendolo con due oggetti GlobalKTable.<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/fc4d91bbe062ceb94f5650224840b81e.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nNe abbiamo gi\u00e0 discusso in precedenza, quindi non mi ripeter\u00f2. Tuttavia, voglio notare che il codice nella funzione toStream().map \u00e8 astratto in un oggetto-funzione invece che in un'espressione lambda inline, per una maggiore leggibilit\u00e0.<\/p>\n<p>Il passo successivo \u00e8 dichiarare due istanze di GlobalKTable (il codice riportato si trova nel file src\/main\/java\/bbejeck\/chapter_5\/GlobalKTableExample.java) (listing 5.12).<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/db3918895f174c8cfb5fc927b55f87a1.jpg\" style=\"display:block;margin: 0 auto;\" \/><\/p>\n<p>Si noti che i nomi dei topic vengono descritti utilizzando tipi enumerati.<\/p>\n<p>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).<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/799360cc99f1920c190a61fd4685d4ff.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nSebbene in questo codice ci siano due connessioni, sono organizzate in forma di catena, poich\u00e9 nessuno dei loro risultati viene utilizzato separatamente. I risultati vengono restituiti alla fine dell'intera operazione.<\/p>\n<p>Eseguendo l'operazione di connessione sopra descritta, otterrete risultati del seguente tipo:<\/p>\n<pre><code class=\"plaintext\">{customer='Barney, Smith' company=\"Exxon\", transactions= 17}<\/code><\/pre>\n<p>\nIl contenuto \u00e8 rimasto invariato, ma questi risultati sono pi\u00f9 chiari.<\/p>\n<p>Se si considera il capitolo 4, \u00e8 gi\u00e0 stato possibile osservare diversi tipi di connessioni in azione. Sono elencati nella tab. 5.2. Questa tabella riflette le possibilit\u00e0 di connessione attuali per la versione 1.0.0 di Kafka Streams; nelle versioni future potrebbero esserci delle modifiche.<\/p>\n<p><img decoding=\"async\" alt=\"Il libro \u00abKafka Streams in azione. Applicazioni e microservizi per il lavoro in tempo reale\u00bb\" src=\"\/wp-content\/uploads\/8e4cf35c64a8431bda43a5e748de275f.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nIn conclusione, ricordo l'aspetto principale: \u00e8 possibile connettere flussi di eventi (KStream) e flussi di aggiornamenti (KTable) utilizzando uno stato locale. Inoltre, se la dimensione dei dati di riferimento non \u00e8 troppo grande, \u00e8 possibile utilizzare l'oggetto GlobalKTable. Il GlobalKTable replica tutte le partizioni su ciascuno dei nodi dell'applicazione Kafka Streams, garantendo cos\u00ec la disponibilit\u00e0 di tutti i dati, indipendentemente dalla partizione a cui appartiene la chiave.<\/p>\n<p>Vedremo quindi una possibilit\u00e0 di Kafka Streams che permette di osservare le modifiche allo stato senza consumare dati dal topic Kafka.<\/p>\n<h3>5.3.5. Stato disponibile per le query<\/h3>\n<p>\nAbbiamo gi\u00e0 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, \u00e8 necessario utilizzare un consumatore Kafka per visualizzarli.<\/p>\n<p>La lettura dei dati da questi argomenti pu\u00f2 essere considerata una forma di viste materializzate (materialized views). Per le nostre necessit\u00e0, possiamo utilizzare la definizione di vista materializzata dall'\u00abWikipedia\u00bb: \u00ab...oggetto fisico di un database che contiene i risultati di un'istruzione. Ad esempio, pu\u00f2 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\u00bb (https:\/\/en.wikipedia.org\/wiki\/Materialized_view).<\/p>\n<p>Kafka Streams consente anche di eseguire query interattive (interactive queries) sugli archivi di stato, il che offre la possibilit\u00e0 di leggere direttamente queste viste materializzate. \u00c8 importante notare che la query sull'archivio di stato \u00e8 un'operazione \u00absolo lettura\u00bb. Grazie a ci\u00f2, puoi evitare di preoccuparti di rendere inconsistente lo stato durante l'elaborazione dei dati da parte dell'applicazione.<\/p>\n<p>La capacit\u00e0 di effettuare query dirette sugli archivi di stato \u00e8 di grande importanza. Ci\u00f2 significa che \u00e8 possibile creare applicazioni \u2014 cruscotti senza la necessit\u00e0 di prima ottenere i dati dal consumatore Kafka. Aumenta anche l'efficienza dell'applicazione, poich\u00e9 non \u00e8 necessario riscrivere nuovamente i dati:<\/p>\n<ul>\n<li>grazie alla localit\u00e0 dei dati, possono essere consultati rapidamente;<\/li>\n<li>si evita la duplicazione dei dati, poich\u00e9 non vengono scritti in un archivio esterno.<\/li>\n<\/ul>\n<p>\nLa cosa principale che vorrei che ricordaste: \u00e8 possibile eseguire direttamente query sullo stato dall'applicazione. Non si pu\u00f2 sottovalutare le opportunit\u00e0 che questo offre. Invece di consumare dati da Kafka e archiviare registrazioni in un database per l'applicazione, \u00e8 possibile effettuare query sugli archivi di stato con lo stesso risultato. Le query dirette sugli archivi di stato significano una minore quantit\u00e0 di codice (assenza di consumatore) e meno software (non \u00e8 necessaria una tabella di database per memorizzare i risultati).<\/p>\n<p>Abbiamo coperto una notevole quantit\u00e0 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\u00e0 di essere integrate nelle applicazioni Kafka Streams, utilizzeremo alcuni degli esempi di questo e del capitolo precedente.<\/p>\n<h3>Riepilogo<\/h3>\n<p><\/p>\n<ul>\n<li>Gli oggetti KStream incarnano flussi di eventi, paragonabili alle inserzioni in un database. Gli oggetti KTable incarnano flussi di aggiornamenti, sono pi\u00f9 simili agli aggiornamenti in un database. La dimensione di un oggetto KTable non cresce, le registrazioni vecchie vengono sostituite da nuove.<\/li>\n<li>Gli oggetti KTable sono necessari per le operazioni di aggregazione.<\/li>\n<li>Le operazioni di finestra possono suddividere i dati aggregati in intervalli temporali.<\/li>\n<li>Grazie agli oggetti GlobalKTable, \u00e8 possibile accedere ai dati di riferimento in qualsiasi punto dell'applicazione, indipendentemente dalla partizione per sezioni.<\/li>\n<li>\u00c8 possibile effettuare join tra gli oggetti KStream, KTable e GlobalKTable.<\/li>\n<\/ul>\n<p>\nFinora 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\u00f9 lunghi di quanto siano stati finora, ma avremo la possibilit\u00e0 di creare praticamente qualsiasi nodo di elaborazione di cui potremmo aver bisogno.<\/p>\n<p>\u2192 Maggiori dettagli sul libro possono essere trovati su <noindex><a rel=\"nofollow\" href=\"https:\/\/www.piter.com\/collection\/best\/product\/kafka-streams-v-deystvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni\">sito dell'editore<\/a><\/noindex><\/p>\n<p>\u2192 Per gli utenti di Habr, uno sconto del 25% con il coupon \u2014 <b>Kafka Streams<\/b><\/p>\n<p>\u2192 Dopo il pagamento della versione cartacea del libro, verr\u00e0 inviata una copia elettronica all'email.<br \/>\n<br \/>Fonte: <a content=\"nofollow\" rel=\"nofollow\" href=\"https:\/\/habr.com\/ru\/company\/piter\/blog\/457756\/\">habr.com<\/a><\/p>","protected":false,"gt_translate_keys":[{"key":"rendered","format":"html"}]},"excerpt":{"rendered":"<p>\u041f\u0440\u0438\u0432\u0435\u0442, \u0425\u0430\u0431\u0440\u043e\u0436\u0438\u0442\u0435\u043b\u0438! \u042d\u0442\u0430 \u043a\u043d\u0438\u0433\u0430 \u043f\u043e\u0434\u043e\u0439\u0434\u0435\u0442 \u0434\u043b\u044f \u043b\u044e\u0431\u043e\u0433\u043e \u0440\u0430\u0437\u0440\u0430\u0431\u043e\u0442\u0447\u0438\u043a\u0430, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0445\u043e\u0447\u0435\u0442 \u0440\u0430\u0437\u043e\u0431\u0440\u0430\u0442\u044c\u0441\u044f \u0432 \u043f\u043e\u0442\u043e\u043a\u043e\u0432\u043e\u0439 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0435. \u041f\u043e\u043d\u0438\u043c\u0430\u043d\u0438\u0435 \u0440\u0430\u0441\u043f\u0440\u0435\u0434\u0435\u043b\u0435\u043d\u043d\u043e\u0433\u043e \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u044f \u043f\u043e\u043c\u043e\u0436\u0435\u0442 \u043b\u0443\u0447\u0448\u0435 \u0438\u0437\u0443\u0447\u0438\u0442\u044c Kafka \u0438 Kafka Streams. \u0411\u044b\u043b\u043e \u0431\u044b \u043d\u0435\u043f\u043b\u043e\u0445\u043e \u0437\u043d\u0430\u0442\u044c \u0438 \u0441\u0430\u043c \u0444\u0440\u0435\u0439\u043c\u0432\u043e\u0440\u043a Kafka, \u043d\u043e \u044d\u0442\u043e \u043d\u0435 \u043e\u0431\u044f\u0437\u0430\u0442\u0435\u043b\u044c\u043d\u043e: \u044f \u0440\u0430\u0441\u0441\u043a\u0430\u0436\u0443 \u0432\u0430\u043c \u0432\u0441\u0435, \u0447\u0442\u043e \u043d\u0443\u0436\u043d\u043e. \u041e\u043f\u044b\u0442\u043d\u044b\u0435 \u0440\u0430\u0437\u0440\u0430\u0431\u043e\u0442\u0447\u0438\u043a\u0438 Kafka, \u043a\u0430\u043a \u0438 \u043d\u043e\u0432\u0438\u0447\u043a\u0438, \u0431\u043b\u0430\u0433\u043e\u0434\u0430\u0440\u044f \u044d\u0442\u043e\u0439 \u043a\u043d\u0438\u0433\u0435 \u043e\u0441\u0432\u043e\u044f\u0442 \u0441\u043e\u0437\u0434\u0430\u043d\u0438\u0435 \u0438\u043d\u0442\u0435\u0440\u0435\u0441\u043d\u044b\u0445 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0439 [&hellip;]<\/p>\n","protected":false,"gt_translate_keys":[{"key":"rendered","format":"html"}]},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[688],"tags":[],"class_list":["post-35786","post","type-post","status-publish","format-standard","hentry","category-administrirovanie"],"aioseo_notices":[],"aioseo_head":"\n\t\t<!-- All in One SEO 5.0.1.1 - aioseo.com -->\n\t<meta name=\"robots\" content=\"max-image-preview:large\" \/>\n\t<meta name=\"author\" content=\"Yuri Gagarin\"\/>\n\t<link rel=\"canonical\" href=\"https:\/\/prohoster.info\/it\/blog\/administrirovanie\/kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni\" \/>\n\t<meta name=\"generator\" content=\"All in One SEO (AIOSEO) 5.0.1.1\" \/>\n\t\t<meta property=\"og:locale\" content=\"it_IT\" \/>\n\t\t<meta property=\"og:site_name\" content=\"ProHoster | \u041a\u0443\u043f\u0438\u0442\u044c \u043d\u0430\u0434\u0435\u0436\u043d\u044b\u0439 \u0445\u043e\u0441\u0442\u0438\u043d\u0433 \u0434\u043b\u044f \u0441\u0430\u0439\u0442\u043e\u0432 \u0441 \u0437\u0430\u0449\u0438\u0442\u043e\u0439 \u043e\u0442 DDoS, VPS VDS \u0441\u0435\u0440\u0432\u0435\u0440\u044b\" \/>\n\t\t<meta property=\"og:type\" content=\"article\" \/>\n\t\t<meta property=\"og:title\" content=\"\ud83e\udd47\u041a\u043d\u0438\u0433\u0430 \u00abKafka Streams \u0432 \u0434\u0435\u0439\u0441\u0442\u0432\u0438\u0438. \u041f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f \u0438 \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u044b \u0434\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b \u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u043c \u0432\u0440\u0435\u043c\u0435\u043d\u0438\u00bb | ProHoster\" \/>\n\t\t<meta property=\"og:url\" content=\"https:\/\/prohoster.info\/it\/blog\/administrirovanie\/kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni\" \/>\n\t\t<meta property=\"og:image\" content=\"https:\/\/prohoster.info\/wp-content\/uploads\/2021\/11\/logo-350.jpg\" \/>\n\t\t<meta property=\"og:image:secure_url\" content=\"https:\/\/prohoster.info\/wp-content\/uploads\/2021\/11\/logo-350.jpg\" \/>\n\t\t<meta property=\"og:image:width\" content=\"350\" \/>\n\t\t<meta property=\"og:image:height\" content=\"350\" \/>\n\t\t<meta property=\"article:published_time\" content=\"2019-10-31T19:06:19+00:00\" \/>\n\t\t<meta property=\"article:modified_time\" content=\"2019-10-31T19:06:19+00:00\" \/>\n\t\t<meta property=\"article:publisher\" content=\"https:\/\/www.facebook.com\/prohoster\" \/>\n\t\t<meta property=\"article:author\" content=\"https:\/\/www.facebook.com\/prohoster\" \/>\n\t\t<!-- All in One SEO -->\n\n","aioseo_head_json":{"title":"\ud83e\udd47Libro \u00abKafka Streams in azione. Applicazioni e microservizi per lavorare in tempo reale\u00bb - ProHoster","description":"","canonical_url":"https:\/\/prohoster.info\/it\/blog\/administrirovanie\/kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni","robots":"max-image-preview:large","keywords":"","webmasterTools":{"miscellaneous":""},"schema":null,"og:locale":"it_IT","og:site_name":"ProHoster | \u041a\u0443\u043f\u0438\u0442\u044c \u043d\u0430\u0434\u0435\u0436\u043d\u044b\u0439 \u0445\u043e\u0441\u0442\u0438\u043d\u0433 \u0434\u043b\u044f \u0441\u0430\u0439\u0442\u043e\u0432 \u0441 \u0437\u0430\u0449\u0438\u0442\u043e\u0439 \u043e\u0442 DDoS, VPS VDS \u0441\u0435\u0440\u0432\u0435\u0440\u044b","og:type":"article","og:title":"\ud83e\udd47\u041a\u043d\u0438\u0433\u0430 \u00abKafka Streams \u0432 \u0434\u0435\u0439\u0441\u0442\u0432\u0438\u0438. \u041f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f \u0438 \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u044b \u0434\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b \u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u043c \u0432\u0440\u0435\u043c\u0435\u043d\u0438\u00bb | ProHoster","og:url":"https:\/\/prohoster.info\/it\/blog\/administrirovanie\/kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni","og:image":"https:\/\/prohoster.info\/wp-content\/uploads\/2021\/11\/logo-350.jpg","og:image:secure_url":"https:\/\/prohoster.info\/wp-content\/uploads\/2021\/11\/logo-350.jpg","og:image:width":350,"og:image:height":350,"article:published_time":"2019-10-31T19:06:19+00:00","article:modified_time":"2019-10-31T19:06:19+00:00","article:publisher":"https:\/\/www.facebook.com\/prohoster","article:author":"https:\/\/www.facebook.com\/prohoster"},"aioseo_meta_data":{"post_id":"35786","title":null,"description":null,"keywords":null,"keyphrases":null,"primary_term":null,"canonical_url":null,"og_title":null,"og_description":null,"og_object_type":"default","og_image_type":"default","og_image_url":null,"og_image_width":null,"og_image_height":null,"og_image_custom_url":null,"og_image_custom_fields":null,"og_video":null,"og_custom_url":null,"og_article_section":null,"og_article_tags":null,"twitter_use_og":false,"twitter_card":"default","twitter_image_type":"default","twitter_image_url":null,"twitter_image_custom_url":null,"twitter_image_custom_fields":null,"twitter_title":null,"twitter_description":null,"schema":{"blockGraphs":[],"customGraphs":[],"default":{"data":{"Article":[],"Course":[],"Dataset":[],"FAQPage":[],"Movie":[],"Person":[],"Product":[],"ProductReview":[],"Car":[],"Recipe":[],"Service":[],"SoftwareApplication":[],"WebPage":[]},"graphName":"","isEnabled":true},"graphs":[]},"schema_type":null,"schema_type_options":null,"pillar_content":false,"robots_default":true,"robots_noindex":false,"robots_noarchive":false,"robots_nosnippet":false,"robots_nofollow":false,"robots_noimageindex":false,"robots_noodp":false,"robots_notranslate":false,"robots_max_snippet":null,"robots_max_videopreview":null,"robots_max_imagepreview":"large","priority":null,"frequency":null,"local_seo":null,"seo_analyzer_scan_date":"2026-01-22 00:45:19","breadcrumb_settings":null,"limit_modified_date":false,"reviewed_by":null,"ai":null,"created":"2021-03-01 01:56:32","updated":"2026-01-22 00:45:19","focus_keyword":null,"additional_keywords":null,"truseo_locale":null},"gt_translate_keys":[{"key":"link","format":"url"}],"_links":{"self":[{"href":"https:\/\/prohoster.info\/it\/wp-json\/wp\/v2\/posts\/35786","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/prohoster.info\/it\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/prohoster.info\/it\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/prohoster.info\/it\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/prohoster.info\/it\/wp-json\/wp\/v2\/comments?post=35786"}],"version-history":[{"count":0,"href":"https:\/\/prohoster.info\/it\/wp-json\/wp\/v2\/posts\/35786\/revisions"}],"wp:attachment":[{"href":"https:\/\/prohoster.info\/it\/wp-json\/wp\/v2\/media?parent=35786"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/prohoster.info\/it\/wp-json\/wp\/v2\/categories?post=35786"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/prohoster.info\/it\/wp-json\/wp\/v2\/tags?post=35786"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}