How would you design an analytics/metrics pipeline?

// Stream aggregation in miniature: 1-minute event-time windows, watermarks, late events, retry duplicates,
// processing-time windows for contrast, and the batch recompute that gives the true answer.
let seed = 11;
function rand() { seed = (seed * 48271) % 2147483647; return seed / 2147483647; }
const W = 60, SECONDS = 600;

const events = [];
for (let i = 0; i < SECONDS * 10; i++) {
  const eventTime = Math.floor(i / 10), r = rand();
  const delay = r < 0.9 ? Math.floor(rand() * 3) : r < 0.98 ? 2 + Math.floor(rand() * 28) : 30 + Math.floor(rand() * 270);
  events.push({ id: "e" + i, eventTime, arrival: eventTime + delay, seq: events.length });
  if (rand() < 0.03) events.push({ id: "e" + i, eventTime, arrival: eventTime + delay + 1 + Math.floor(rand() * 5), seq: events.length }); // producer retry
}
const log = events.slice().sort((a, b) => a.arrival - b.arrival || a.seq - b.seq); // the order the stream sees
const dups = events.length - SECONDS * 10;
console.log(`${SECONDS * 10} real events over ${SECONDS} s of event time, ${dups} retry duplicates, ${events.length} messages in the log`);

// Batch truth: distinct event ids per event-time window, computed over the complete log.
const truth = Array(SECONDS / W).fill(0), seenAll = new Set();
for (const e of log) if (!seenAll.has(e.id)) { seenAll.add(e.id); truth[Math.floor(e.eventTime / W)]++; }

function stream(lag, dedup) {
  const counts = new Map(), emitted = new Map(), seen = new Set();
  let watermark = -Infinity, maxEvent = -Infinity, late = 0, latency = 0, closedByWm = 0, dupCounted = 0;
  const ids = new Set();
  const flush = (now, final) => {
    for (const [w, c] of [...counts.entries()].sort((a, b) => a[0] - b[0])) {
      if (watermark >= (w + 1) * W) { emitted.set(w, c); counts.delete(w); if (!final) { latency += now - (w + 1) * W; closedByWm++; } }
    }
  };
  for (const e of log) {
    if (dedup) { if (seen.has(e.id)) continue; seen.add(e.id); }
    const w = Math.floor(e.eventTime / W);
    if (emitted.has(w)) { late++; continue; } // window already closed and emitted
    if (ids.has(e.id)) dupCounted++; else ids.add(e.id);
    counts.set(w, (counts.get(w) || 0) + 1);
    maxEvent = Math.max(maxEvent, e.eventTime);
    watermark = maxEvent - lag;
    flush(e.arrival, false);
  }
  const lastArrival = log[log.length - 1].arrival;
  watermark = Infinity; flush(lastArrival, true); // end of input closes the rest
  let err = 0; truth.forEach((t, w) => (err += Math.abs((emitted.get(w) || 0) - t)));
  return { late, err, dupCounted, closedByWm, latency: latency / closedByWm, emitted };
}

console.log("\nwatermark lag  dedup  duplicates counted  late dropped  total error vs batch  emit delay after window end");
for (const [lag, dedup] of [[5, false], [5, true], [30, true], [120, true], [300, true]]) {
  const r = stream(lag, dedup);
  console.log(`${String(lag).padStart(9)} s  ${String(dedup).padEnd(5)}  ${String(r.dupCounted).padStart(18)}  ${String(r.late).padStart(12)}  ${String(r.err).padStart(20)}  ${r.latency.toFixed(1).padStart(7)} s avg over ${r.closedByWm} windows`);
}

const procTime = Array(SECONDS / W).fill(0), seenP = new Set();
for (const e of log) { if (seenP.has(e.id)) continue; seenP.add(e.id); const w = Math.floor(e.arrival / W); if (w < procTime.length) procTime[w]++; }
const r30 = stream(30, true);
console.log("\nwindow       batch truth  event-time (lag 30 s)  processing-time");
truth.forEach((t, w) => console.log(`${String(w * W).padStart(3)}-${String((w + 1) * W).padEnd(3)} s ${String(t).padStart(12)} ${String(r30.emitted.get(w)).padStart(22)} ${String(procTime[w]).padStart(16)}`));

console.log("\n== Capacity (assumptions in, arithmetic out) ==");
const perDay = 2e9, bytes = 500, peakFactor = 3, partitionMBs = 5, columnarRatio = 10;
const avg = perDay / 86400, peak = avg * peakFactor, peakMBs = (peak * bytes) / 1e6;
console.log(`events/s: average ${Math.round(avg)}, peak ${Math.round(peak)} (x${peakFactor})`);
console.log(`peak ingest ${peakMBs.toFixed(1)} MB/s -> at least ${Math.ceil(peakMBs / partitionMBs)} partitions at ${partitionMBs} MB/s each`);
console.log(`raw log ${(perDay * bytes) / 1e12} TB/day; columnar at ${columnarRatio}:1 ${(perDay * bytes) / columnarRatio / 1e9} GB/day`);
console.log(`minute rollup upper bound: 1440 min x 50 countries x 3 devices x 1000 pages = ${1440 * 50 * 3 * 1000} rows/day`);

// Distinct counts in an analytics pipeline: exact sets vs HyperLogLog sketches, and why
// hourly unique counts cannot be added up but hourly sketches can be merged.
let seed = 5;
function rand() { seed = (seed * 48271) % 2147483647; return seed / 2147483647; }
function hash32(s) { // FNV-1a + murmur3 finalizer
  let h = 0x811c9dc5;
  for (let i = 0; i < s.length; i++) { h ^= s.charCodeAt(i); h = Math.imul(h, 16777619) >>> 0; }
  h ^= h >>> 16; h = Math.imul(h, 0x85ebca6b) >>> 0; h ^= h >>> 13; h = Math.imul(h, 0xc2b2ae35) >>> 0; h ^= h >>> 16;
  return h >>> 0;
}
class HLL {
  constructor(p) { this.p = p; this.m = 1 << p; this.reg = new Uint8Array(this.m); }
  add(s) {
    const h = hash32(s), idx = h >>> (32 - this.p), w = (h << this.p) >>> 0;
    const rho = w === 0 ? 32 - this.p + 1 : Math.clz32(w) + 1;
    if (rho > this.reg[idx]) this.reg[idx] = rho;
  }
  merge(o) { for (let i = 0; i < this.m; i++) if (o.reg[i] > this.reg[i]) this.reg[i] = o.reg[i]; }
  count() {
    let z = 0, zeros = 0;
    for (let i = 0; i < this.m; i++) { z += Math.pow(2, -this.reg[i]); if (this.reg[i] === 0) zeros++; }
    const alpha = 0.7213 / (1 + 1.079 / this.m);
    let e = (alpha * this.m * this.m) / z;
    if (e <= 2.5 * this.m && zeros > 0) e = this.m * Math.log(this.m / zeros); // small-range correction
    return Math.round(e);
  }
  bytes() { return (this.m * 6) / 8; } // 6-bit registers when packed
}
const USERS = 100000;
const day10 = new HLL(10), day14 = new HLL(14), hourly = [];
for (let h = 0; h < 24; h++) hourly.push({ exact: new Set(), sketch: new HLL(14) });
const dayExact = new Set();
let events = 0;
for (let u = 0; u < USERS; u++) {
  const id = "user-" + u, n = 1 + Math.floor(rand() * 6);
  for (let k = 0; k < n; k++) {
    const hour = Math.floor(rand() * 24); events++;
    dayExact.add(id); day10.add(id); day14.add(id);
    hourly[hour].exact.add(id); hourly[hour].sketch.add(id);
  }
}
const errPct = (est, truth) => (((est - truth) * 100) / truth).toFixed(2) + "%";
console.log(`${events} events from ${dayExact.size} distinct users in one day`);
console.log("\nmethod                 memory      estimate   error    expected std error");
const row = (label, mem, est, err, std) => console.log(`${label.padEnd(23)}${mem.padEnd(12)}${String(est).padStart(8)}   ${err.padStart(6)}    ${std}`);
row("exact set", "every id", dayExact.size, "0.00%", "-");
for (const s of [day10, day14]) {
  const est = s.count();
  row(`HLL p=${s.p} (${s.m} regs)`, `${s.bytes()} B`, est, errPct(est, dayExact.size), ((1.04 / Math.sqrt(s.m)) * 100).toFixed(2) + "%");
}
const sumHourly = hourly.reduce((a, h) => a + h.exact.size, 0);
const merged = new HLL(14); for (const h of hourly) merged.merge(h.sketch);
console.log("\n== Rolling 24 hourly results up to a daily number ==");
console.log(`hour 0 exact uniques: ${hourly[0].exact.size}, hour 0 sketch: ${hourly[0].sketch.count()}`);
console.log(`sum of 24 hourly exact unique counts: ${sumHourly} (${(sumHourly / dayExact.size).toFixed(2)}x the true ${dayExact.size})`);
const m = merged.count();
console.log(`merge of 24 hourly sketches (register max): ${m} (${errPct(m, dayExact.size)})`);