import json, time, random

import pandas as pd

Progetto: pipeline real-time end-to-end

Questo lab mette insieme tutto il modulo in un unico progetto: un’azienda che vuole passare da report giornalieri a una pipeline che collega eventi, stream processor, ClickHouse, viste aggregate e dashboard operative. Ogni passaggio introduce un rischio diverso, dai duplicati ai ritardi, dai costi alle definizioni instabili fino agli alert inutili, e il lavoro consiste nel tenere insieme questi trade-off in una decisione architetturale completa.

Il problema da risolvere

Il punto non è conoscere la pipeline in astratto. È decidere cosa fare quando il team ha dati incompleti, metriche ambigue o vincoli tecnici che rendono fragile la lettura del fenomeno. Una pipeline utile separa il segnale dal rumore, dichiara quale baseline usa e rende più difendibile l’azione che segue.

Leggi il caso come una design review. Ogni componente deve avere un motivo, un owner e un criterio di fallimento. La domanda non è “possiamo farlo in real-time?”, ma quali decisioni migliorano abbastanza da giustificare la complessità, il monitoraggio e il costo operativo aggiuntivo.

Come impostare il lavoro

Conviene procedere per passaggi, dove ognuno chiarisce il costo di una decisione sbagliata.

Passaggio	Domanda da fare	Output atteso
Decisione	Che cosa cambia se la pipeline funziona?	Scelta esplicita
Segnale	Quale dato osservabile riduce l’incertezza?	Metrica o evento
Baseline	Rispetto a cosa interpretiamo il risultato?	Confronto credibile
Vincolo	Che cosa può falsare la lettura?	Assunzione da dichiarare
Azione	Quale passo operativo segue?	Raccomandazione controllabile

In un progetto reale la pipeline non vive isolata. È parte di un sistema più ampio fatto di decisioni, dati disponibili, vincoli tecnici, incentivi organizzativi e qualità dell’esecuzione. Il rischio dell’analista alle prime armi è trattare il tema come una definizione da memorizzare. Il lavoro professionale è capire quale problema risolve, quali assunzioni contiene e cosa succede quando quelle assunzioni saltano. La prima domanda non è “quale metrica calcolo?” ma quale decisione dovrà essere presa grazie a questa analisi. Una dashboard o una query hanno valore solo se riducono incertezza decisionale; se non cambiano una scelta, sono documentazione o teatro analitico.

Formalizzare la decisione

Per rendere il progetto analizzabile conviene definire prima l’unità di lavoro, che qui può essere un evento, una finestra temporale, una materialized view, un alert o una metrica live. Quella unità va poi collegata a un segnale osservabile come latenza, freshness, falsi positivi, throughput o costo query. Infine si dichiara la decisione attesa, che riguarda la scelta tra pipeline realtime, vista aggregata, alert o dashboard operativa.

Elemento	Definizione operativa	Controllo minimo
Unità di analisi	Oggetto su cui misuri il fenomeno	Utente, account, evento, ordine o periodo
Variabile osservata	Segnale che rappresenta il comportamento	Definizione stabile e tracciabile
Baseline	Stato contro cui confronti il segnale	Periodo, segmento, controllo o benchmark
Soglia decisionale	Punto in cui cambia l’azione	Criterio scritto prima della lettura
Rischio residuo	Errore che può restare dopo l’analisi	Sensitivity check o revisione qualitativa

La formalizzazione non complica la lezione. Rende visibili le assunzioni, così uno stakeholder può discutere il criterio decisionale invece di fidarsi del risultato per autorità. È solida quando un altro analista riproduce la logica, critica le assunzioni e arriva alla stessa decisione partendo dagli stessi dati.

Come leggere i segnali

Immagina un team che deve usare la pipeline per decidere se cambiare una metrica, un investimento o una dashboard. La domanda non è quale sia la definizione corretta, ma quale scelta diventa meno rischiosa se l’analisi è fatta bene.

Situazione	Lettura prudente	Decisione
Il dato migliora ma la baseline è debole	Il segnale potrebbe dipendere dal campione	Rafforzare il confronto prima di scalare
La metrica cambia in un solo segmento	L’effetto medio nasconde eterogeneità	Separare coorti o casi d’uso
Il costo operativo aumenta	Il beneficio va valutato sul margine	Applicare una soglia economica esplicita
Il sistema produce numeri incoerenti	La fiducia nel dato è parte della decisione	Correggere ownership e controlli

Fase 1: setup Kafka e ClickHouse (30 min)

docker-compose up -d kafka clickhouse
# Verifica: docker-compose logs kafka | grep "started"

Fase 2: producer Python (20 min)

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8'))

while True:
    event = {"user_id": random.randint(1,1000), "event": random.choice(
        ["page_view","add_cart","purchase"]), "amount": round(random.uniform(10,200),2),
        "timestamp": time.time()}
    producer.send('ecommerce_events', event)
    time.sleep(0.1)  # 10 msg/sec

Fase 3: tabelle ClickHouse (20 min)

CREATE TABLE kafka_events (...) ENGINE = Kafka SETTINGS ...;
CREATE MATERIALIZED VIEW mv_events_per_min
ENGINE = SummingMergeTree() ORDER BY (minute, event)
AS SELECT toStartOfMinute(toDateTime(timestamp)) AS minute,
       event, count() AS cnt, sum(amount) AS revenue
FROM kafka_events GROUP BY minute, event;

Fase 4: dashboard con Grafana (20 min)

Crea tre pannelli. Il primo mostra gli ordini al minuto, come time series su mv_events_per_min filtrato per event='purchase'. Il secondo mostra il revenue al minuto, come time series su SUM(revenue). Il terzo costruisce il conversion funnel page_view → add_cart → purchase con i tassi di conversione tra uno stadio e l’altro.

Fase 5: alert di anomalia (15 min)

Configura un alert in Grafana: se il tasso di purchase scende sotto la baseline del 50%, invia una notifica Slack.

Consegna

• Producer Python in esecuzione (≥10 msg/sec)
• ClickHouse con MV popolata in tempo reale
• Dashboard Grafana con 3 pannelli funzionanti
• Alert configurato e testato

Caso reale: Netflix e la disciplina delle metriche

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 che il dato non viene trattato come ornamento, bensì come infrastruttura decisionale.

Quando Netflix valuta una modifica all’esperienza, che sia una nuova riga di raccomandazioni, una diversa immagine di copertina o un 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: la pipeline deve essere collegata a un outcome, e se il risultato non aiuta a scegliere tra due azioni alternative l’analisi è incompleta.

Esempio SQL: una vista di controllo

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;

Questa query non è la risposta finale. Crea una superficie di osservazione fatta di trend, segmenti, differenze tra canali e variazioni nel tempo, da cui l’analista formula ipotesi più precise.

Esempio Python: stabilità e anomalie

Una metrica utile deve essere abbastanza stabile da orientare decisioni e abbastanza sensibile da segnalare cambiamenti reali. In Python possiamo controllare 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 di questo controllo è pratico: evita di reagire a ogni oscillazione casuale e segnala quando una variazione merita un’indagine. In un contesto aziendale questo tipo di analisi alimenta alert, review settimanali e retrospettive di prodotto.

Errori comuni

Il primo errore è lavorare su dati aggregati troppo presto, perché una media globale può nascondere due segmenti che si muovono in direzioni opposte. Il secondo è non controllare la qualità del dato: eventi duplicati, tracking incompleto, timezone incoerenti e cambi di definizione producono conclusioni false. Il terzo è confondere correlazione e causalità. Se gli utenti che usano una feature convertono di più, non significa che la feature causi la conversione: potrebbero usarla perché erano già più motivati. Per ridurre questi rischi, ogni analisi dovrebbe includere almeno tre controlli: una definizione esplicita della metrica, un confronto per segmento e una verifica contro un periodo precedente o un gruppo di controllo.

Esercizio guidato

Al livello base, scrivi in cinque righe quale decisione concreta la pipeline dovrebbe migliorare, 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: segnale osservato, interpretazione prudente, controllo necessario e azione consigliata. Usa almeno un caso in cui il segnale da solo non basta per decidere.

Al livello research-grade trasforma l’esercizio in un memo decisionale che includa assunzioni, criteri di esclusione, soglia di intervento, sensitivity check e una proposta di monitoraggio dopo la decisione. Come dataset usa un export reale, una tabella sintetica o un notebook di studio, purché contenga almeno una dimensione di segmento, una metrica osservabile e un periodo di confronto.

Checkpoint

1. Quale decisione concreta dovrebbe migliorare questa pipeline?
2. Quale unità di analisi rende il problema misurabile?
3. Quale baseline useresti per evitare una lettura isolata?
4. Quale assunzione, se falsa, cambierebbe la conclusione?
5. Quale controllo presenteresti prima di raccomandare un’azione?

Riepilogo

La pipeline real-time è uno strumento decisionale, non un argomento da manuale. Il valore nasce quando colleghi problema, dati, metrica, segmentazione e azione. Una buona analisi non termina con “il numero è salito” o “il numero è sceso”, ma con una frase operativa: quale decisione prendiamo, con quale livello di confidenza, e quale metrica useremo per sapere se avevamo ragione. La forma corretta resta sempre la stessa, fatta di decisione, segnale, baseline, rischio e azione; tutto il resto serve solo se rende più affidabile uno di questi passaggi.