08 - Game Telemetry Pipeline: Player Analytics su Scala

Telemetria di gioco: dall'evento che parte dal client alla decisione sul prodotto

Il giorno in cui qualcuno ti chiede «perché i giocatori lasciano dopo il terzo livello?», la risposta sta in una tabella che avresti dovuto costruire mesi prima. La telemetria è la raccolta degli eventi di gioco (sessioni, partite, acquisti, errori) in un posto dove si possano contare. Costruirla bene è meno questione di strumenti e più di tre difetti che quasi tutte le prime versioni hanno: eventi duplicati, eventi che arrivano in ritardo e segmenti calcolati con una query che sembra giusta e non lo è.

L'articolo segue il percorso dell'evento: la busta con cui nasce, il client che lo spedisce, Kafka che lo conserva, lo stream processing che lo aggrega e il database analitico che lo interroga. Ogni parte che si può eseguire in questa macchina è un listato con l'uscita reale: sono modelli in un processo con dati sintetici e seme fisso, quindi dimostrano un comportamento e non misurano il tuo traffico. Kafka, Flink e ClickHouse non sono installati qui: per quelli mi appoggio alla documentazione ufficiale e dichiaro quando un frammento non è stato eseguito.

I dati che alimentano gli eventi live si leggono con le tecniche di LiveOps con feature flag ed esperimenti, e il monitoraggio tecnico del backend (latenza, tick) è nell'articolo Observability di un game backend. Qui parliamo di dati sui giocatori; quelli sulle attese e sugli abbandoni servono anche a valutare la qualità del matchmaking con Elo e Glicko-2.

La busta dell'evento

Ogni evento ha la stessa busta e un contenuto che dipende dal tipo. La busta è ciò che permette di trattarli in massa: senza di essa nessuno sa come deduplicare, ordinare o unire gli eventi di una sessione. Un esempio concreto: un gameplay.kill avvenuto durante la partita match_789xyz porta nella busta chi l'ha fatto, quando, in quale build e su quale piattaforma, e nel contenuto chi è stato colpito, con quale arma e a che distanza.

• event_id: un identificativo unico generato dal client, che consente di riconoscere un duplicato
• event_type e schema_version: che cosa è successo e quale forma ha il contenuto, per far convivere versioni diverse
• player_id, session_id, match_id: a chi, in quale sessione e in quale partita si riferisce
• client_ts e server_ts: l'orologio del dispositivo e quello del server
• game_build e platform: per attribuire un problema a una versione o a una piattaforma

Delle due date, quella del server è la fonte di verità: l'orologio di un telefono può essere sbagliato di ore, e un giocatore può cambiarlo apposta. Il client_ts si conserva perché la differenza fra i due dice quanto era lenta la consegna, un dato utile per i ritardi che incontrerai fra poco. Il campo event_id è il più importante dell'elenco per la qualità dei dati: senza di esso ogni ritentativo di rete diventa un evento in più.

Il client: lotti, ritentativi e il caso della conferma persa

Spedire un evento per volta costa una richiesta di rete ciascuno, quindi il client li accumula e li invia a lotti, con un limite di tempo e uno di dimensione. Il problema serio è il ritentativo. Se il client invia un lotto e la risposta non torna, non può sapere se il server l'ha salvato o no. Ci sono due guasti diversi: la richiesta non è arrivata, oppure è arrivata e a perdersi è stata la conferma. Nel secondo caso l'unico comportamento sicuro per il client è rimandare il lotto, ed è esattamente ciò che crea i duplicati.

Il listato simula entrambe le situazioni con uno schema di esiti fisso (ok, ack_lost, request_lost). Il Gateway ricorda gli event_id visti per un periodo e scarta i ripetuti entro quel periodo; è la stessa idea di uno stato con TTL in un job di streaming. L'Sdk tiene il lotto in testa al buffer finché non riceve una conferma. Nota la riga con ack_lost: il gateway ha già chiamato accept, e il metodo ritorna comunque false.

interface TEvent { event_id: string; event_type: string; client_ts: number; }
type Outcome = 'ok' | 'request_lost' | 'ack_lost';

const HOUR = 3_600_000;

// Gateway: salva gli eventi nuovi e ricorda gli event_id visti per una finestra di tempo (come lo stato con TTL di un job di streaming)
class Gateway {
  seen: Map<string, number> = new Map();
  stored: TEvent[] = [];
  received = 0;
  ttlMs: number;
  constructor(ttlMs: number) { this.ttlMs = ttlMs; }
  accept(batch: TEvent[], nowMs: number): void {
    for (const e of batch) {
      this.received++;
      const first = this.seen.get(e.event_id);
      if (first !== undefined && nowMs - first < this.ttlMs) continue;   // duplicato dentro la finestra: scartato
      this.seen.set(e.event_id, nowMs);
      this.stored.push(e);
    }
  }
}

// SDK: accumula, invia a lotti, e se non riceve conferma tiene il lotto e lo rimanda
class Sdk {
  buffer: TEvent[] = [];
  gateway: Gateway;
  outcomes: Outcome[];
  attempt = 0;
  constructor(gateway: Gateway, outcomes: Outcome[]) { this.gateway = gateway; this.outcomes = outcomes; }
  track(type: string, id: number, ts: number): void { this.buffer.push({ event_id: 'ev-' + id, event_type: type, client_ts: ts }); }
  flush(nowMs: number, batchSize: number): boolean {
    const batch = this.buffer.slice(0, batchSize);
    if (batch.length === 0) return true;
    const outcome = this.outcomes[this.attempt++ % this.outcomes.length];
    if (outcome === 'request_lost') return false;                     // il gateway non ha visto nulla
    this.gateway.accept(batch, nowMs);                                // il gateway ha salvato...
    if (outcome === 'ack_lost') return false;                         // ...ma la conferma non è tornata: per l'SDK è un errore
    this.buffer.splice(0, batch.length);
    return true;
  }
}

La seconda metà del file esegue lo scenario. Con 20 eventi e un esito su cinque che perde la conferma, il gateway riceve 35 eventi in totale e ne salva esattamente 20: nessuna perdita e nessun duplicato. Poi un telefono rimasto offline rimanda un lotto dopo 25 ore, oltre la finestra di 24 ore del gateway: l'evento ev-3 viene salvato una seconda volta. Non è un difetto del listato, è il costo di una finestra finita: la memoria per ricordare tutti gli identificativi cresce con il tempo, e la finestra decide quanti ritardi potrai perdonare. Scegli il valore guardando quanto a lungo resta offline il tuo client più sfortunato, e prevedi una deduplicazione a valle, nel database analitico, per quel che sfugge.

const gw = new Gateway(24 * HOUR);
const sdk = new Sdk(gw, ['ok', 'ack_lost', 'request_lost', 'ok', 'ack_lost']);
for (let i = 0; i < 20; i++) sdk.track('gameplay.kill', i, i * 1000);
let t = 0;
while (sdk.buffer.length > 0) { sdk.flush(t, 5); t += 1000; }
console.log(`inviati 20 eventi: il gateway ne ha ricevuti ${gw.received}, ne ha salvati ${gw.stored.length}`);

// Un telefono resta offline: il lotto salvato su disco viene rimandato 25 ore dopo
const offline = new Sdk(gw, ['ok']);
offline.track('economy.purchase', 3, 3000);                           // stesso event_id 'ev-3' già salvato
offline.flush(25 * HOUR, 5);
console.log(`dopo il rinvio a 25 ore: salvati ${gw.stored.length} (ev-3 compare ${gw.stored.filter(e => e.event_id === 'ev-3').length} volte)`);

// Output:
// inviati 20 eventi: il gateway ne ha ricevuti 35, ne ha salvati 20
// dopo il rinvio a 25 ore: salvati 21 (ev-3 compare 2 volte)

Kafka offre anche un produttore idempotente: la documentazione dice che dalla 0.11.0.0 il produttore può garantire che un rinvio non produca duplicati nel log. Ma quella garanzia copre la tratta fra il tuo gateway e Kafka, non quella fra il telefono e il gateway: i due duplicati vanno trattati separatamente.

Kafka: che cosa conserva, e una configurazione che sembra prudente

Kafka è il registro durevole in mezzo alla pipeline: gli eventi si scrivono in topic, divisi in partizioni, e più consumatori li rileggono a ritmo proprio. Scegliere la chiave del messaggio decide l'ordinamento: la documentazione spiega che il produttore sceglie la partizione con una funzione sulla chiave, e che se la chiave è un identificativo utente, tutti i dati di quell'utente finiscono nella stessa partizione. Con player_id come chiave, gli eventi di un giocatore restano in ordine fra loro.

La trappola è nella politica di pulizia. Con cleanup.policy=compact Kafka mantiene, per ogni chiave, l'ultimo valore: lo dice la pagina del design, «will always retain at least the last known value for each message key». Su un topic di eventi chiavato per giocatore ciò significa che la storia dei suoi eventi viene ridotta all'ultimo. È utile per uno stato (il profilo corrente), è distruttivo per un registro di eventi. Per la telemetria si resta sulla politica delete, che scarta i segmenti vecchi alla scadenza.

# Schema da adattare, NON eseguito: Kafka non è installato sulla macchina di lavoro.
# Topic degli eventi di gameplay, chiave del messaggio = player_id (ordine per giocatore).
kafka-topics.sh --create --bootstrap-server kafka:9092 \
  --topic game.gameplay.events \
  --partitions 12 --replication-factor 3 \
  --config cleanup.policy=delete \
  --config retention.ms=2592000000

Non ho potuto eseguirlo: Kafka non è installato in questa macchina, e il comando segue la sintassi della guida del progetto. Il numero di partizioni è il massimo di parallelismo dei consumatori, non un'ottimizzazione che si cambia a cuor leggero: aumentarlo dopo cambia la partizione in cui finiscono i messaggi nuovi di una stessa chiave, e può rompere l'ordine fra vecchi e nuovi. Il valore 12 è un esempio da dimensionare sul tuo traffico, non una raccomandazione.

Il tempo dell'evento e i ritardatari

Un motore di stream processing come Flink aggrega per finestre di tempo. Quale tempo? Se conta l'istante in cui l'evento arriva, una connessione lenta sposta un kill nel minuto sbagliato. Se conta l'istante in cui l'evento è accaduto, bisogna decidere quando una finestra è completa, e qui entra in gioco il watermark: un segnale che dichiara «non aspetto più eventi più vecchi di questo». Con forBoundedOutOfOrderness il watermark resta indietro rispetto al massimo tempo visto di un margine fisso. Un evento che arriva quando la sua finestra è già chiusa è «tardivo»: secondo la documentazione di Flink viene scartato per impostazione predefinita, perché la tolleranza aggiuntiva (allowedLateness) vale zero, e si può invece recuperare con un'uscita laterale (sideOutputLateData).

Il listato non usa Flink: riproduce la regola. Genera 3.000 eventi in 10 minuti, il 3% dei quali arriva da connessioni lente con 60-120 secondi di ritardo, e conta quanti vengono scartati al variare del margine. La scelta del margine è un compromesso fra completezza e attesa, perché il risultato di una finestra non esce prima di quel margine dopo la sua fine.

import random

# Modello didattico del tempo evento: 3000 eventi in 10 minuti, che arrivano in ritardo e fuori ordine.
rng = random.Random(7)
events = []
for i in range(3000):
    t = rng.uniform(0, 600)                        # istante in cui è successo (secondi)
    slow = rng.random() < 0.03                     # il 3% arriva da connessioni lente: 60-120 s di ritardo
    delay = rng.uniform(60, 120) if slow else rng.expovariate(1 / 4)   # gli altri: media 4 s
    events.append((t + delay, t))                  # (istante di arrivo, istante dell'evento)
events.sort()                                      # il motore li vede in ordine di arrivo

def run(bound_s: float):
    """Finestre da 60 s; la finestra si chiude quando watermark = max(istante evento) - bound supera la sua fine."""
    max_ts, closed, counted, dropped = -1.0, set(), 0, 0
    for _arrival, ts in events:
        w = int(ts // 60)
        if w in closed:
            dropped += 1                           # finestra già emessa: l'evento è tardivo e viene scartato
        else:
            counted += 1
        max_ts = max(max_ts, ts)
        watermark = max_ts - bound_s
        for done in range(int(watermark // 60)):   # chiude le finestre il cui termine è <= watermark
            closed.add(done)
    return counted, dropped

print("bound  contati  scartati  quota scartata  attesa minima del risultato")
for bound in (0, 10, 30, 120):
    c, d = run(bound)
    print(f"{bound:>4} s {c:>8} {d:>9} {100 * d / len(events):>13.1f}%  {bound:>10} s dopo la fine della finestra")

# Output:
# bound  contati  scartati  quota scartata  attesa minima del risultato
#    0 s     2803       197           6.6%           0 s dopo la fine della finestra
#   10 s     2913        87           2.9%          10 s dopo la fine della finestra
#   30 s     2932        68           2.3%          30 s dopo la fine della finestra
#  120 s     3000         0           0.0%         120 s dopo la fine della finestra

Con margine zero si perde il 6,6% degli eventi; con 10 secondi il 2,9%; con 30 secondi il 2,3%; solo a 120 secondi, oltre il ritardo massimo generato, non si perde nulla, e per averli bisogna aspettare due minuti dopo la fine di ogni finestra. Osserva il passo da 10 a 30: guadagni poco, perché il 3% di ritardatari lenti è fuori portata di entrambi. Per quella coda lunga non serve un margine enorme: serve la corsia laterale, che raccoglie i tardivi e li corregge in un secondo momento. Le percentuali dipendono da come ho generato i ritardi, non dal tuo gioco: misura la distribuzione di server_ts - client_ts sui tuoi dati prima di scegliere il numero.

Segmentare i giocatori con RFM, e il quintile che mente

RFM (Recency, Frequency, Monetary) assegna a ogni giocatore tre punteggi: quanto tempo è passato dall'ultima sessione, quante sessioni ha fatto di recente, quanto ha speso. Dalla combinazione dei tre nascono i segmenti su cui agisce il LiveOps (chi sta per andarsene, chi spende molto). L'implementazione abituale divide ogni misura in cinque gruppi con NTILE(5). Il rischio è che la funzione divide le righe in gruppi di dimensione uguale e non guarda i valori: secondo la documentazione di SQLite «divide la partizione in N gruppi il più uniformi possibile».

Nel listato, 1.000 giocatori sintetici, di cui il 12% ha mai pagato. La query è eseguita su SQLite 3.46.1, che supporta le funzioni finestra; ClickHouse documenta anche ntile ma con un dialetto diverso, e non l'ho eseguito. Il primo risultato è quello che colpisce: dei 897 giocatori con spesa zero, 97 ricevono il punteggio monetario 5, il più alto.

import random, sqlite3

# 1000 giocatori sintetici. Eventi: session_start e purchase. "Oggi" è il giorno 100.
rng = random.Random(11)
db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE events (player_id INTEGER, event_type TEXT, day INTEGER, amount REAL)")
rows = []
for p in range(1000):
    last_active = 100 - int(rng.expovariate(1 / 12))
    for _ in range(rng.randint(1, 40)):
        rows.append((p, "session_start", max(0, last_active - int(rng.expovariate(1 / 15))), 0.0))
    if rng.random() < 0.12:                              # solo il 12% ha mai pagato
        for _ in range(rng.randint(1, 4)):
            rows.append((p, "purchase", max(0, last_active - rng.randint(0, 80)), round(rng.uniform(2, 60), 2)))
db.executemany("INSERT INTO events VALUES (?,?,?,?)", rows)

BASE = """
WITH base AS (
  SELECT player_id,
         100 - MAX(CASE WHEN event_type = 'session_start' THEN day END)                    AS recency_days,
         SUM(event_type = 'session_start' AND day > 70)                                   AS frequency_30d,
         COALESCE(SUM(CASE WHEN event_type = 'purchase' AND day > 10 THEN amount END), 0) AS monetary_90d
  FROM events GROUP BY player_id)
"""
NAIVE = BASE + """
SELECT m_score, COUNT(*) AS giocatori_con_spesa_zero FROM (
  SELECT monetary_90d, NTILE(5) OVER (ORDER BY monetary_90d) AS m_score FROM base)
WHERE monetary_90d = 0 GROUP BY m_score ORDER BY m_score"""
FIXED = BASE + """
SELECT m_score, COUNT(*) AS giocatori FROM (
  SELECT CASE WHEN monetary_90d = 0 THEN 0
              ELSE NTILE(5) OVER (PARTITION BY monetary_90d > 0 ORDER BY monetary_90d) END AS m_score
  FROM base)
GROUP BY m_score ORDER BY m_score"""

print("NTILE(5) sulla spesa, solo i giocatori con spesa 0:")
for r in db.execute(NAIVE): print("  m_score", r[0], "->", r[1], "giocatori")
print("spesa 0 separata, quintili solo tra chi ha pagato:")
for r in db.execute(FIXED): print("  m_score", r[0], "->", r[1], "giocatori")

# Output:
# NTILE(5) sulla spesa, solo i giocatori con spesa 0:
#   m_score 1 -> 200 giocatori
#   m_score 2 -> 200 giocatori
#   m_score 3 -> 200 giocatori
#   m_score 4 -> 200 giocatori
#   m_score 5 -> 97 giocatori
# spesa 0 separata, quintili solo tra chi ha pagato:
#   m_score 0 -> 897 giocatori
#   m_score 1 -> 21 giocatori
#   m_score 2 -> 21 giocatori
#   m_score 3 -> 21 giocatori
#   m_score 4 -> 20 giocatori
#   m_score 5 -> 20 giocatori

Con una spesa uguale a zero per quasi tutti, NTILE deve comunque riempire cinque gruppi, e li riempie con persone che hanno speso lo stesso importo: lo zero finisce spezzato fra i punteggi da 1 a 5 per l'ordine in cui le righe capitano. Un giocatore che non ha mai pagato può risultare un «big spender». La correzione, nella seconda query, è separare chi ha speso zero (punteggio 0) e calcolare i quintili solo fra chi ha pagato: così 897 giocatori valgono 0 e i 103 paganti si distribuiscono in gruppi da circa 20. La regola generale: ogni volta che molti valori sono uguali, i quantili per rango non dicono nulla di quel gruppo. Controlla sempre la distribuzione prima di dare un nome al segmento.

Che cosa farne dei segmenti

Un segmento è utile solo se ha un'azione associata e un modo di misurarne l'effetto. «Giocatori a rischio» con un invito a tornare si valuta con un esperimento, come nell'articolo su LiveOps, tenendo un gruppo di controllo a cui non scrivi. Le percentuali «tipiche» di ciascun segmento che si trovano in giro non hanno una fonte: il tuo gioco ha la sua distribuzione, ed è la prima cosa da guardare.

Feature per prevedere l'abbandono

Prevedere l'abbandono richiede un modello e dati etichettati («ha smesso di giocare nei 7 giorni successivi»). La pipeline di telemetria non fa questo: fornisce le feature, cioè misure per giocatore su una finestra mobile. Nel listato la funzione riceve gli eventi di un giocatore e la fine della finestra, e calcola sessioni, durata media, quota di partite abbandonate, spesa e giorni dall'ultima sessione sui 7 giorni precedenti.

from dataclasses import dataclass

@dataclass(frozen=True)
class Ev:
    player: str
    kind: str        # session_start | session_end | match_end | match_abandon | purchase
    ts_ms: int       # tempo dell'evento (server)
    value: float = 0.0

DAY = 86_400_000

def features(events: list[Ev], window_end_ms: int) -> dict:
    """Feature di un giocatore sulla finestra [fine-7gg, fine). Il riferimento è la FINE DELLA FINESTRA, non l'orologio."""
    win = [e for e in events if window_end_ms - 7 * DAY <= e.ts_ms < window_end_ms]
    count = lambda k: sum(1 for e in win if e.kind == k)
    ended, abandoned = count("match_end"), count("match_abandon")
    starts = [e.ts_ms for e in win if e.kind == "session_start"]
    durs = [e.value for e in win if e.kind == "session_end"]
    return {
        "sessions_7d": count("session_start"),
        "avg_session_min": round(sum(durs) / len(durs) / 60, 1) if durs else 0.0,
        "abandon_rate": round(abandoned / (ended + abandoned), 3) if ended + abandoned else 0.0,
        "spend_7d": round(sum(e.value for e in win if e.kind == "purchase"), 2),
        "days_since_last_session": (window_end_ms - max(starts)) // DAY if starts else None,   # None = nessuna sessione nella finestra
    }

END = 30 * DAY
evs = [Ev("p1", "session_start", END - 5 * DAY), Ev("p1", "session_end", END - 5 * DAY + 1, 1500),
       Ev("p1", "match_end", END - 5 * DAY + 2), Ev("p1", "match_abandon", END - 4 * DAY),
       Ev("p1", "purchase", END - 4 * DAY, 4.99), Ev("p1", "session_start", END - 8 * DAY)]   # quest'ultimo è fuori finestra
print(features(evs, END))
print("stesso input, stesso risultato:", features(evs, END) == features(evs, END))
print("stessa funzione, finestra spostata di 3 giorni:", features(evs, END + 3 * DAY))

# Output:
# {'sessions_7d': 1, 'avg_session_min': 25.0, 'abandon_rate': 0.5, 'spend_7d': 4.99, 'days_since_last_session': 5}
# stesso input, stesso risultato: True
# stessa funzione, finestra spostata di 3 giorni: {'sessions_7d': 0, 'avg_session_min': 0.0, 'abandon_rate': 1.0, 'spend_7d': 4.99, 'days_since_last_session': None}

La scelta di progetto è nell'argomento window_end_ms: i «giorni dall'ultima sessione» si misurano rispetto alla fine della finestra e non all'orologio di sistema. Con l'orologio, ricalcolare la stessa finestra domani darebbe un numero diverso, e il modello verrebbe allenato su dati che nessuno può riprodurre. Il listato dimostra il caso al contrario: stesso input, stesso risultato (True). L'uscita mostra anche un caso limite da gestire: se non ci sono sessioni nella finestra il valore è None e non un numero inventato come 999. Quanto sia affidabile un modello costruito su queste feature non lo dice questo codice: va misurato su dati reali con una prova a posteriori.

Quando tutto questo è troppo

Kafka, Flink e un database analitico sono tre sistemi da far funzionare. Per uno studio con pochi giocatori, una tabella di eventi in PostgreSQL con una chiave unica su event_id risolve i duplicati e risponde alle stesse domande. Il passaggio a una pipeline distribuita ha senso quando la scrittura singola non regge il volume o quando servono aggregazioni in tempo reale. Prima di quel punto, le parti più preziose sono comunque la busta, la deduplicazione e il tempo evento: valgono con qualsiasi strumento sotto.

Domande frequenti

Che cosa sai fare adesso

Sai progettare la busta di un evento e spiegare perché serve un identificativo unico. Sai prevedere i duplicati dovuti alle conferme perse e dimensionare la finestra di deduplicazione, e sai perché la compattazione di Kafka non va sui topic di eventi. Sai scegliere un margine di fuori-ordine misurando la coda dei ritardi, e riconoscere un quintile costruito su valori uguali. Resta da misurare, sui tuoi dati reali, la distribuzione dei ritardi e quella della spesa: sono i due numeri che il codice di questo articolo non può darti.