import pandas as pd

Kafka Streams: processing events with Java

Prima o poi un’applicazione deve arricchire eventi, calcolare aggregati e reagire a sequenze di comportamento senza uscire dall’ecosistema Kafka. Kafka Streams porta questa logica dentro un’applicazione deployabile, ma con un costo: introduce stato locale, changelog, repartition e una serie di failure mode propri. Conviene quindi leggerlo non come una libreria di trasformazioni, ma come la progettazione di un servizio stateful, in cui la domanda giusta non è quale metrica calcolo, ma quale decisione cambia se l’applicazione resta corretta anche mentre scala o riparte.

Perché lo stato cambia tutto

Una topology Kafka Streams si valuta seguendo lo stato: dove nasce, dove viene salvato, come si ricostruisce e cosa succede durante un rebalance. È corretta quando conserva semantica, ordine e recuperabilità anche mentre l’applicazione cambia numero di istanze o riparte dopo un crash. Tre vincoli decidono il disegno: quanto stato locale serve, quanta retention tenere sui changelog e come gestire i late events. Se ignori uno di questi tre punti, l’applicazione funziona in demo e si rompe in produzione.

The Kafka Streams model

A stream processing in Kafka Streams always follows this pattern:

Source Topic → Stream Processor → Sink Topic

Per esempio, per filtrare le transazioni sospette bastano poche righe:

KStream<String, Transaction> transactions = builder.stream("transactions");
KStream<String, Transaction> suspicious = transactions
    .filter((key, txn) -> txn.getAmount() > 10000 && txn.getCountry().equals("NG"));
suspicious.to("suspicious_transactions");

Leggi, filtri, scrivi. Sotto questa apparente semplicità, Kafka Streams gestisce automaticamente lo stato, il ribilanciamento tra istanze e la tolleranza ai guasti. È proprio questa automazione che rende il sistema potente e, allo stesso tempo, difficile da debuggare quando qualcosa va storto.

Stateful operations: aggregations and windows

Le operazioni interessanti sono quelle che mantengono stato. Per contare gli eventi per tipo ogni 5 minuti:

KTable<Windowed<String>, Long> counts = transactions
    .groupBy((key, txn) -> txn.getType())
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
    .count();

Per arricchire ogni evento con dati esterni si usa il join tra stream e tabella:

KStream<String, Transaction> enriched = transactions
    .join(userTable, (txn, user) ->
        new EnrichedTransaction(txn, user.getName()));

Il join unisce ogni transazione con i dati utente più recenti, dove la tabella è un log compatto di Kafka. È il pattern che rende possibile arricchire eventi in tempo reale senza interrogare un database esterno per ogni messaggio.

Exactly-once semantics

Kafka Streams implementa l’exactly-once nativamente dal 2017. Usa le transazioni Kafka per scrivere atomicamente su più topic, garantendo che ogni evento venga processato esattamente una volta anche in caso di crash e riavvio. Si attiva con processing.guarantee=exactly_once. La garanzia non è gratis (costa latenza e throughput), quindi va richiesta solo quando la duplicazione di un evento avrebbe conseguenze reali, come in un conteggio finanziario.

Kafka Streams o Flink

La scelta tra Kafka Streams e Flink dipende da quanto è pesante l’elaborazione e da quanta infrastruttura vuoi gestire.

	Kafka Streams	Flink
Deployment	Embedded library	Separate cluster
Complexity	Simple, native Java/Kafka	Complex, requires infrastructure
Ideal use case	Lightweight transformations, enrichment	Heavy aggregations, ML on stream, complex event time
Maximum throughput	Millions/msg sec	Tens of millions/msg sec

In pratica Kafka Streams vince quando vuoi trasformazioni leggere e arricchimento senza un cluster da mantenere, mentre Flink vale la complessità in più quando hai aggregazioni pesanti, machine learning sullo stream o una gestione del tempo evento complessa.

Esempio: scegliere se usare Kafka Streams

Immagina un team che vuole calcolare sessioni utente e alert comportamentali direttamente dallo stream. Prima di scegliere Kafka Streams deve valutare la dimensione dello stato, la retention dei changelog, la gestione dei late events e l’impatto dei rebalancing, perché l’applicazione diventa a tutti gli effetti parte della piattaforma dati. La tabella aiuta a leggere i segnali tipici.

Observed evidence	Cautious interpretation	Recommended action
Il throughput migliora	Potrebbe essere un effetto reale o una variazione normale di carico	Cercare un confronto e un segmento
Una istanza accumula più stato delle altre	La distribuzione delle chiavi è sbilanciata	Rivedere la chiave o il numero di partizioni
Il costo cresce insieme allo stato	L’impatto va letto sul margine, non sul totale	Stimare il trade-off di retention dei changelog

Real case: Netflix and the discipline of metrics

Netflix è un esempio utile perché ha costruito molte decisioni di prodotto intorno a segnali comportamentali osservabili: completamento degli episodi, tempo di ricerca prima della riproduzione, abbandono dopo pochi minuti, ritorno nella settimana successiva, efficacia delle raccomandazioni. Il punto non è che ogni azienda debba copiare Netflix, ma il metodo: il dato non viene trattato come ornamento, viene trattato come infrastruttura decisionale.

Quando Netflix valuta una modifica all’esperienza, per esempio una nuova riga di raccomandazioni o un diverso algoritmo di ranking, non misura solo il click immediato. Misura anche segnali di qualità: l’utente guarda davvero il contenuto, torna nei giorni successivi, riduce il tempo speso a cercare. Questa disciplina impedisce di ottimizzare vanity metric che sembrano positive nel breve periodo ma erodono valore nel lungo. Lo stesso vale qui: una pipeline Kafka Streams ha senso solo se il suo output cambia una decisione.

SQL example: building a control view

Il pattern seguente è volutamente generico ma eseguibile nella maggior parte dei warehouse moderni. Serve a creare una base analitica con metrica, segmento e finestra temporale, così da confrontare periodi e gruppi senza riscrivere la logica ogni volta.

WITH base_events AS (
  SELECT
    user_id,
    account_id,
    event_type,
    event_time,
    DATE_TRUNC('week', event_time) AS week,
    source,
    device_type
  FROM events
  WHERE event_time >= CURRENT_DATE - INTERVAL '180 days'
    AND user_id IS NOT NULL
),
weekly_user_metrics AS (
  SELECT
    week,
    user_id,
    COALESCE(source, 'unknown') AS source,
    COALESCE(device_type, 'unknown') AS device_type,
    COUNT(*) AS total_events,
    COUNT(DISTINCT DATE(event_time)) AS active_days,
    COUNT(DISTINCT event_type) AS event_diversity,
    MAX(CASE WHEN event_type IN ('purchase', 'subscribe', 'activation') THEN 1 ELSE 0 END) AS reached_key_outcome
  FROM base_events
  GROUP BY week, user_id, source, device_type
)
SELECT
  week,
  source,
  device_type,
  COUNT(DISTINCT user_id) AS users,
  ROUND(AVG(active_days), 2) AS avg_active_days,
  ROUND(AVG(event_diversity), 2) AS avg_event_diversity,
  ROUND(AVG(reached_key_outcome) * 100, 2) AS key_outcome_rate
FROM weekly_user_metrics
GROUP BY week, source, device_type
ORDER BY week, source, device_type;

La query non è la risposta finale. Crea una superficie di osservazione (trend, segmenti, differenze tra canali, variazioni nel tempo) da cui formulare ipotesi più precise.

Python example: checking stability and anomalies

Una metrica utile deve essere abbastanza stabile da orientare decisioni e abbastanza sensibile da segnalare cambiamenti reali. In Python puoi controllare le variazioni anomale settimana su settimana con un z-score mobile.

# df contiene: week, segment, users, key_outcome_rate
# key_outcome_rate espresso in percentuale, es. 12.4

df = df.sort_values(['segment', 'week']).copy()
df['previous_rate'] = df.groupby('segment')['key_outcome_rate'].shift(1)
df['wow_change_pp'] = df['key_outcome_rate'] - df['previous_rate']
df['rolling_mean'] = df.groupby('segment')['key_outcome_rate'].transform(
    lambda s: s.rolling(4, min_periods=2).mean()
)
df['rolling_std'] = df.groupby('segment')['key_outcome_rate'].transform(
    lambda s: s.rolling(4, min_periods=2).std()
)
df['z_score'] = (df['key_outcome_rate'] - df['rolling_mean']) / df['rolling_std']

anomalies = df[df['z_score'].abs() >= 2].sort_values('z_score')
print(anomalies[['week', 'segment', 'key_outcome_rate', 'wow_change_pp', 'z_score']])

Il valore di questo controllo è pratico. Evita di reagire a ogni oscillazione casuale e segnala solo quando una variazione merita un’indagine, alimentando alert, review settimanali e retrospettive di prodotto.

Errori tipici da evitare

L’errore più frequente è usare Kafka Streams come etichetta invece che come processo. Succede quando si mostra una pipeline senza dire quale decisione produce, una metrica senza baseline o una conclusione senza indicare quale assunzione potrebbe invalidarla. La domanda di controllo è semplice: se questo risultato fosse instabile, quale scelta sbaglierei?

Sul lato dati ci sono tre trappole ricorrenti. La prima è lavorare su aggregati troppo presto, perché una media globale nasconde segmenti che si muovono in direzioni opposte. La seconda è non controllare la qualità del dato: eventi duplicati, tracking incompleto, timezone incoerenti e cambi di definizione producono conclusioni false. La terza è confondere correlazione e causalità, dato che gli utenti che usano una feature potrebbero convertire di più solo perché erano già più motivati. Tre controlli minimi riducono il rischio: definizione esplicita della metrica, confronto per segmento e verifica contro un periodo precedente.

Lab ed esercizi

Per consolidare, lavora su tre livelli.

1. Scrivi in cinque righe quale decisione concreta deve supportare la tua topology Kafka Streams, indicando metrica, unità di analisi, baseline e rischio principale.
2. Costruisci una tabella con quattro colonne (segnale osservato, interpretazione prudente, controllo necessario, azione consigliata) e includi almeno un caso in cui il segnale non basta per decidere.
3. Trasforma l’esercizio in un decision memo con assunzioni, criteri di esclusione, soglia di intervento, sensitivity check e un piano di monitoraggio dopo la decisione.

Come materiale puoi usare Kafka, schema registry, log eventi, consumer lag e un dataset clickstream. In assenza di dati reali, crea un dataset sintetico con almeno 200 righe, una dimensione temporale, una di segmento e una metrica di outcome.

Checkpoint

Prima di chiudere, verifica di saper rispondere. Quale decisione concreta deve migliorare la pipeline? Quale unità di analisi rende il problema misurabile? Quale baseline useresti per evitare una lettura ingenua? Quale errore tipico potrebbe cambiare la conclusione? Quale output consegneresti a uno stakeholder non tecnico?

---

References:

• Confluent. (2024). “Kafka Streams Documentation.” kafka.apache.org/streams.
• Stopford, B. (2018). Designing Event-Driven Systems. O’Reilly.

Operational Summary

Kafka Streams è utile quando riduce l’incertezza su una scelta reale, non quando aggiunge un layer di astrazione. La progettazione corretta segue lo stato dell’applicazione: dove vive, come si ricostruisce, cosa succede durante un rebalance. Una pipeline ben fatta non finisce con “il numero è salito”: finisce con una decisione, un livello di confidenza e una metrica che ti dirà se avevi ragione.