import pandas as pd

Project: end-to-end Kafka pipeline

Questo progetto mette insieme sorgenti, topic, schema, consumer, sink e monitoraggio in un’unica pipeline Kafka funzionante. La sfida non è far passare un messaggio in una demo, cosa che riesce a chiunque, ma garantire che replay, duplicati, evoluzione dello schema e lag restino governati quando la pipeline diventa una dipendenza reale del business. È un lab, quindi il punto non è accumulare definizioni ma arrivare a una pipeline che si possa difendere in una review.

Come leggere il progetto

Conviene affrontarlo come una review architetturale, non come un esercizio di copia e incolla. Ogni topic deve avere un owner, una retention dichiarata, uno schema, un insieme atteso di consumer e un criterio di qualità. Se una parte della pipeline non si riesce a spiegare in termini di responsabilità e di modalità di guasto, non è ancora pronta per la produzione, per quanto bene giri in locale.

La prima domanda da farsi non è “quale metrica calcolo” ma “quale decisione dovrà migliorare grazie a questa pipeline”. Una dashboard o una query hanno valore solo se riducono l’incertezza di una scelta concreta. Se non cambiano nessuna decisione, sono documentazione o teatro tecnico.

Architettura del caso

Il caso ricostruisce il flusso di una piattaforma di food delivery: gli ordini arrivano da un microservizio, vengono arricchiti con lo stato della consegna tramite Kafka Streams e finiscono in un topic da cui leggono i consumer analitici. La decisione finale, ossia mandare in produzione o no, dipende dai test di replay, dai contratti di schema, dalla gestione dei duplicati e dalla presenza di un runbook per gli incidenti.

Prima di procedere conviene fissare l’unità di lavoro (topic, evento, schema, producer, consumer o stream processor), il segnale che osservi (latenza, throughput, lag, compatibilità schema, perdita dati), la baseline rispetto a cui lo leggi e la decisione attesa. Il rischio costante è scambiare un numero disponibile per una prova sufficiente: una pipeline che gira non è ancora una pipeline affidabile.

Fase 1: setup e topic (20 min)

docker-compose up -d kafka zookeeper schema-registry
kafka-topics --create --topic orders --partitions 8 --replication-factor 1
kafka-topics --create --topic deliveries --partitions 8
kafka-topics --create --topic enriched_orders --partitions 8

Il numero di partizioni va deciso ora, perché aumentarlo dopo ridistribuisce le chiavi e può rompere le garanzie di ordine su cui contano i consumer.

Fase 2: producer Python (20 min)

# Simulate orders from the orders microservice
for i in range(1000):
    order = {"order_id": i, "restaurant_id": random.randint(1,50),
             "amount": round(random.uniform(10,100),2),
             "timestamp": time.time()}
    producer.produce('orders', key=str(order['order_id']),
                     value=json.dumps(order))

La chiave dell’ordine determina la partizione, quindi tutti gli eventi di uno stesso ordine restano ordinati tra loro, che è ciò che serve per arricchirli correttamente più avanti.

Phase 3: Kafka Streams (30 min)

// Enrich orders with delivery status
KStream<String, Order> orders = builder.stream("orders");
KStream<String, Delivery> deliveries = builder.stream("deliveries");
orders.join(deliveries, (order, delivery) ->
    new EnrichedOrder(order, delivery.getStatus()),
    JoinWindows.of(Duration.ofMinutes(30)))
.to("enriched_orders");

La finestra di join di 30 minuti è la decisione progettuale più delicata: troppo stretta e perdi gli abbinamenti per le consegne lente, troppo larga e tieni stato in memoria più del necessario.

Fase 4: consumer e monitoring (15 min)

Verifica il consumer lag con kafka-consumer-groups --describe. Un lag che cresce in modo lineare dice che il consumer non recupererà da solo, e va affrontato prima del rilascio.

Delivery

• Order producer working
• Kafka Streams join active
• Consumer lag <1000
• Topic enriched_orders populated

Errori tipici nel progetto

L’errore più comune è trattare la pipeline come una definizione: imparare i nomi dei componenti, ricordare due configurazioni, applicare un template. Il lavoro reale è diverso, perché bisogna capire quale problema risolve ogni pezzo, quali assunzioni contiene e cosa succede quando quelle assunzioni saltano. Una pipeline reale non vive isolata: è dentro un sistema fatto di decisioni, dati disponibili, vincoli tecnici, incentivi organizzativi e qualità dell’esecuzione.

Un secondo errore è presentare un numero senza dire quale decisione cambia, quale baseline lo rende interpretabile e quale rischio resta aperto. In quel caso il dato sembra preciso ma non guida l’azione. La domanda di controllo è semplice: se questo risultato fosse instabile, quale scelta sbaglierei? Se non sai rispondere, manca ancora il collegamento tra analisi e azione.

Esercizi

Al livello base, scrivi in cinque righe quale decisione concreta dovrebbe migliorare grazie a questa pipeline, indicando metrica, unità di analisi, baseline e rischio principale. Se non riesci a nominare la decisione, il progetto è ancora troppo astratto.

Al livello intermedio, costruisci una tabella con quattro colonne, ossia segnale osservato, interpretazione prudente, controllo necessario e azione consigliata, includendo almeno un caso in cui il segnale da solo non basta per decidere.

Al livello research-grade, trasforma l’esercizio in un decision memo che includa assunzioni, criteri di esclusione, soglia di intervento, sensitivity check e un piano di monitoraggio dopo la decisione. Come materiale puoi usare Kafka con lo schema registry, log di eventi, consumer lag e un dataset clickstream; in mancanza di dati reali, costruisci un dataset sintetico di almeno 200 righe con una dimensione temporale, una di segmento e una metrica di outcome.

Real case: Netflix and the discipline of metrics

Netflix è un esempio utile perché ha costruito molte decisioni di prodotto attorno 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 che il dato viene trattato come infrastruttura decisionale e non come ornamento.

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. Guarda anche i segnali di qualità: l’utente guarda davvero il contenuto, torna nei giorni successivi, riduce il tempo speso a cercare. Questa disciplina evita di ottimizzare vanity metric che sembrano positive nel breve periodo ma erodono valore nel lungo. Lo stesso principio vale per la pipeline di questo progetto: il suo output deve essere collegato a un outcome, altrimenti l’analisi è incompleta.

Esempio SQL: una vista di controllo

Il pattern seguente è generico ma eseguibile sulla 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 ogni volta la logica.

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 dà la risposta finale: crea una superficie di osservazione fatta di trend, segmenti e differenze tra canali, da cui formulare ipotesi più precise.

Esempio Python: stabilità e anomalie

Una metrica utile deve essere abbastanza stabile da orientare le decisioni e abbastanza sensibile da segnalare cambiamenti reali. In Python si possono controllare le variazioni anomale settimana su settimana.

# 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 è pratico: il controllo evita di reagire a ogni oscillazione casuale e segnala solo le variazioni che meritano un’indagine. Lo stesso meccanismo può alimentare alert, review settimanali e retrospettive di prodotto.

Summary

Una pipeline Kafka end-to-end va trattata come uno strumento decisionale, non come un argomento da manuale. Il valore nasce quando colleghi problema, dati, metrica, segmentazione e azione, e quando ogni topic ha un owner, una retention e un comportamento prevedibile sotto guasto. Una buona analisi non si chiude con “il numero è salito” o “il numero è sceso”, ma con una frase operativa: quale decisione prendiamo, con quale confidenza, e quale metrica useremo per sapere se avevamo ragione.