08 - Game Telemetry Pipelines: Player Behavior Analytics at Scale

Game telemetry: from the event leaving the client to the product decision

The day someone asks you "why do players leave after the third level?", the answer lives in a table you should have built months earlier. Telemetry is the collection of game events (sessions, matches, purchases, errors) in a place where they can be counted. Building it well is less a matter of tools and more of three defects that almost all first versions have: duplicate events, events that arrive late and segments computed with a query that looks right and is not.

The article follows the path of the event: the envelope it is born with, the client that ships it, Kafka that stores it, the stream processing that aggregates it and the analytical database that queries it. Every part that can be run on this machine is a listing with the real output: they are single-process models with synthetic data and a fixed seed, so they demonstrate a behaviour and do not measure your traffic. Kafka, Flink and ClickHouse are not installed here: for those I lean on the official documentation and I say when a fragment was not run.

The data that feeds live events is read with the techniques of LiveOps with feature flags and experiments, and the technical monitoring of the backend (latency, tick) is in the article Observability of a game backend. Here we talk about player data; data on waiting times and abandonment also serves to evaluate the quality of matchmaking with Elo and Glicko-2.

The event envelope

Every event has the same envelope and a content that depends on the type. The envelope is what lets you process them in bulk: without it nobody knows how to deduplicate, order or join the events of a session. A concrete example: a gameplay.kill that happened during the match match_789xyz carries in the envelope who did it, when, in which build and on which platform, and in the content who was hit, with which weapon and at what distance.

• event_id: a unique identifier generated by the client, which makes it possible to recognise a duplicate
• event_type and schema_version: what happened and what shape the content has, so that different versions can coexist
• player_id, session_id, match_id: to whom, in which session and in which match it refers
• client_ts and server_ts: the device's clock and the server's clock
• game_build and platform: to attribute a problem to a version or a platform

Of the two dates, the server's is the source of truth: a phone's clock can be wrong by hours, and a player can change it on purpose. The client_ts is kept because the difference between the two says how slow the delivery was, a useful figure for the delays you will meet shortly. The event_id field is the most important in the list for data quality: without it every network retry becomes one extra event.

The client: batches, retries and the case of the lost acknowledgement

Sending one event at a time costs one network request each, so the client accumulates them and sends them in batches, with a time limit and a size limit. The serious problem is the retry. If the client sends a batch and the response does not come back, it cannot know whether the server saved it or not. There are two different failures: the request did not arrive, or it arrived and what got lost was the acknowledgement. In the second case the only safe behaviour for the client is to resend the batch, and that is exactly what creates the duplicates.

The listing simulates both situations with a fixed pattern of outcomes (ok, ack_lost, request_lost). The Gateway remembers the event_ids seen for a period and discards the repeated ones within that period; it is the same idea as a state with a TTL in a streaming job. The Sdk keeps the batch at the head of the buffer until it receives an acknowledgement. Note the line with ack_lost: the gateway has already called accept, and the method returns false anyway.

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

const HOUR = 3_600_000;

// Gateway: stores new events and remembers the event_ids seen for a time window (like the TTL state of a streaming job)
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;   // duplicate inside the window: discarded
      this.seen.set(e.event_id, nowMs);
      this.stored.push(e);
    }
  }
}

// SDK: accumulates, sends in batches, and if it gets no acknowledgement keeps the batch and resends it
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;                     // the gateway saw nothing
    this.gateway.accept(batch, nowMs);                                // the gateway stored it...
    if (outcome === 'ack_lost') return false;                         // ...but the acknowledgement did not come back: for the SDK it is an error
    this.buffer.splice(0, batch.length);
    return true;
  }
}

The second half of the file runs the scenario. With 20 events and one outcome in five that loses the acknowledgement, the gateway receives 35 events in total and stores exactly 20: no loss and no duplicates. Then a phone that stayed offline resends a batch after 25 hours, beyond the gateway's 24-hour window: the event ev-3 is stored a second time. This is not a defect of the listing, it is the cost of a finite window: the memory to remember all the identifiers grows with time, and the window decides how many delays you can forgive. Choose the value by looking at how long your unluckiest client stays offline, and plan a downstream deduplication, in the analytical database, for what slips through.

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(`sent 20 events: the gateway received ${gw.received}, stored ${gw.stored.length}`);

// A phone stays offline: the batch saved on disk is resent 25 hours later
const offline = new Sdk(gw, ['ok']);
offline.track('economy.purchase', 3, 3000);                           // same event_id 'ev-3' already stored
offline.flush(25 * HOUR, 5);
console.log(`after the resend at 25 hours: stored ${gw.stored.length} (ev-3 appears ${gw.stored.filter(e => e.event_id === 'ev-3').length} times)`);

// Output:
// sent 20 events: the gateway received 35, stored 20
// after the resend at 25 hours: stored 21 (ev-3 appears 2 times)

Kafka also offers an idempotent producer: the documentation says that since 0.11.0.0 the producer can guarantee that a resend does not produce duplicates in the log. But that guarantee covers the leg between your gateway and Kafka, not the one between the phone and the gateway: the two kinds of duplicate must be treated separately.

Kafka: what it keeps, and a setting that looks prudent

Kafka is the durable log in the middle of the pipeline: events are written into topics, split into partitions, and several consumers re-read them at their own pace. Choosing the message key decides the ordering: the documentation explains that the producer chooses the partition with a function on the key, and that if the key is a user identifier, all of that user's data ends up in the same partition. With player_id as the key, a player's events stay in order with respect to each other.

The trap is in the cleanup policy. With cleanup.policy=compact Kafka keeps, for each key, the latest value: the design page says so, "will always retain at least the last known value for each message key". On an event topic keyed by player this means that the history of their events is reduced to the last one. It is useful for state (the current profile), and destructive for an event log. For telemetry you stay on the delete policy, which discards old segments at expiry.

# Schema to adapt, NOT executed: Kafka is not installed on the working machine.
# Gameplay events topic, message key = player_id (ordering per player).
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

I could not run it: Kafka is not installed on this machine, and the command follows the syntax of the project's guide. The number of partitions is the maximum consumer parallelism, not an optimisation to be changed lightly: increasing it later changes the partition where new messages of the same key land, and can break the ordering between old and new. The value 12 is an example to be sized on your traffic, not a recommendation.

Event time and the latecomers

A stream processing engine such as Flink aggregates by time windows. Which time? If what counts is the instant the event arrives, a slow connection moves a kill into the wrong minute. If what counts is the instant the event happened, you have to decide when a window is complete, and this is where the watermark comes in: a signal that declares "I will no longer wait for events older than this". With forBoundedOutOfOrderness the watermark stays behind the maximum time seen by a fixed margin. An event that arrives when its window is already closed is "late": according to the Flink documentation it is dropped by default, because the additional tolerance (allowedLateness) is zero, and it can instead be recovered with a side output (sideOutputLateData).

The listing does not use Flink: it reproduces the rule. It generates 3,000 events over 10 minutes, 3% of which arrive from slow connections with 60-120 seconds of delay, and counts how many are dropped as the margin varies. The choice of margin is a compromise between completeness and waiting, because the result of a window does not come out before that margin after its end.

import random

# Teaching model of event time: 3000 events in 10 minutes, arriving late and out of order.
rng = random.Random(7)
events = []
for i in range(3000):
    t = rng.uniform(0, 600)                        # instant at which it happened (seconds)
    slow = rng.random() < 0.03                     # 3% come from slow connections: 60-120 s of delay
    delay = rng.uniform(60, 120) if slow else rng.expovariate(1 / 4)   # the others: mean 4 s
    events.append((t + delay, t))                  # (arrival instant, event instant)
events.sort()                                      # the engine sees them in arrival order

def run(bound_s: float):
    """60 s windows; a window closes when watermark = max(event instant) - bound passes its end."""
    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                           # window already emitted: the event is late and is discarded
        else:
            counted += 1
        max_ts = max(max_ts, ts)
        watermark = max_ts - bound_s
        for done in range(int(watermark // 60)):   # closes the windows whose end is <= watermark
            closed.add(done)
    return counted, dropped

print("bound  counted  dropped  dropped share  minimum wait for the result")
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 after the window ends")

# Output:
# bound  counted  dropped  dropped share  minimum wait for the result
#    0 s     2803       197           6.6%           0 s after the window ends
#   10 s     2913        87           2.9%          10 s after the window ends
#   30 s     2932        68           2.3%          30 s after the window ends
#  120 s     3000         0           0.0%         120 s after the window ends

With a zero margin you lose 6.6% of the events; with 10 seconds, 2.9%; with 30 seconds, 2.3%; only at 120 seconds, beyond the maximum delay generated, do you lose nothing, and to have them all you must wait two minutes after the end of every window. Notice the step from 10 to 30: you gain little, because the 3% of slow latecomers is out of reach of both. For that long tail you do not need a huge margin: you need the side lane, which collects the late events and corrects them later. The percentages depend on how I generated the delays, not on your game: measure the distribution of server_ts - client_ts on your data before choosing the number.

Segmenting players with RFM, and the quintile that lies

RFM (Recency, Frequency, Monetary) assigns each player three scores: how much time has passed since the last session, how many sessions they have played recently, how much they have spent. From the combination of the three come the segments LiveOps acts on (who is about to leave, who spends a lot). The usual implementation divides each measure into five groups with NTILE(5). The risk is that the function divides the rows into groups of equal size and does not look at the values: according to the SQLite documentation it "divides the partition into N groups as evenly as possible".

In the listing, 1,000 synthetic players, of whom 12% have ever paid. The query is run on SQLite 3.46.1, which supports window functions; ClickHouse also documents ntile but with a different dialect, and I did not run it. The first result is the one that strikes: of the 897 players with zero spend, 97 receive the monetary score 5, the highest.

import random, sqlite3

# 1000 synthetic players. Events: session_start and purchase. "Today" is day 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:                              # only 12% have ever paid
        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 players_with_zero_spend 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 players 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) on spend, only the players with spend 0:")
for r in db.execute(NAIVE): print("  m_score", r[0], "->", r[1], "players")
print("zero spend kept apart, quintiles only among those who paid:")
for r in db.execute(FIXED): print("  m_score", r[0], "->", r[1], "players")

# Output:
# NTILE(5) on spend, only the players with spend 0:
#   m_score 1 -> 200 players
#   m_score 2 -> 200 players
#   m_score 3 -> 200 players
#   m_score 4 -> 200 players
#   m_score 5 -> 97 players
# zero spend kept apart, quintiles only among those who paid:
#   m_score 0 -> 897 players
#   m_score 1 -> 21 players
#   m_score 2 -> 21 players
#   m_score 3 -> 21 players
#   m_score 4 -> 20 players
#   m_score 5 -> 20 players

With a spend equal to zero for almost everyone, NTILE must still fill five groups, and it fills them with people who spent the same amount: the zero ends up split among the scores from 1 to 5 by the order in which the rows happen to fall. A player who has never paid can turn out to be a "big spender". The fix, in the second query, is to separate those who spent zero (score 0) and compute the quintiles only among those who paid: this way 897 players get 0 and the 103 payers are spread into groups of about 20. The general rule: whenever many values are equal, rank-based quantiles say nothing about that group. Always check the distribution before giving the segment a name.

What to do with the segments

A segment is useful only if it has an action attached and a way of measuring its effect. "At-risk players" with an invitation to come back is evaluated with an experiment, as in the article on LiveOps, keeping a control group you do not write to. The "typical" percentages for each segment that you find around have no source: your game has its own distribution, and it is the first thing to look at.

Features for predicting churn

Predicting churn requires a model and labelled data ("stopped playing in the following 7 days"). The telemetry pipeline does not do this: it supplies the features, that is per-player measures over a sliding window. In the listing the function receives the events of one player and the end of the window, and computes sessions, average duration, share of abandoned matches, spend and days since the last session over the previous 7 days.

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       # event time (server)
    value: float = 0.0

DAY = 86_400_000

def features(events: list[Ev], window_end_ms: int) -> dict:
    """Features of a player over the window [end-7d, end). The reference is the WINDOW END, not the clock."""
    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 = no session in the window
    }

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)]   # the last one is outside the window
print(features(evs, END))
print("same input, same result:", features(evs, END) == features(evs, END))
print("same function, window shifted by 3 days:", 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}
# same input, same result: True
# same function, window shifted by 3 days: {'sessions_7d': 0, 'avg_session_min': 0.0, 'abandon_rate': 1.0, 'spend_7d': 4.99, 'days_since_last_session': None}

The design choice is in the window_end_ms argument: the "days since the last session" are measured against the end of the window and not the system clock. With the clock, recomputing the same window tomorrow would give a different number, and the model would be trained on data that nobody can reproduce. The listing demonstrates the opposite case: same input, same result (True). The output also shows an edge case to handle: if there are no sessions in the window the value is None and not an invented number like 999. How reliable a model built on these features is, this code does not say: it must be measured on real data with a backtest.

When all this is too much

Kafka, Flink and an analytical database are three systems to keep running. For a studio with few players, an events table in PostgreSQL with a unique key on event_id solves the duplicates and answers the same questions. The move to a distributed pipeline makes sense when a single writer cannot sustain the volume or when real-time aggregations are needed. Before that point, the most valuable parts are still the envelope, the deduplication and the event time: they hold with any tool underneath.

Frequently asked questions

What you can do now

You can design an event envelope and explain why a unique identifier is needed. You can predict the duplicates caused by lost acknowledgements and size the deduplication window, and you know why Kafka's compaction does not belong on event topics. You can choose an out-of-order margin by measuring the tail of the delays, and recognise a quintile built on equal values. What remains is to measure, on your real data, the distribution of delays and that of spend: they are the two numbers that this article's code cannot give you.