Design a Real-Time Event Analytics System

Framing statement:

“Meta needs to query billions of events with sub-second latency for debugging, monitoring, and product analytics. The key challenge is interactive querying at petabyte scale — engineers must be able to slice-and-dice events by any dimension in real-time, unlike traditional batch analytics that take minutes. I’ll focus on the ingestion pipeline, in-memory columnar storage, and query engine.”

Phased delivery:

Team ownership:

Key ambiguity I’d challenge:

“When we say ’real-time,’ do we mean all 10B daily events queryable within 30 seconds? Or is sampling acceptable? Full fidelity needs ~2.5PB in-memory; priority sampling at 10% average brings this to ~250TB. I want to clarify freshness and fidelity expectations per event category.”

Functional Requirements (Phase 1)

• ·FR1: Ingest structured events from any Meta service via Kafka with schema validation
• ·FR2: Events queryable within 30 seconds of emission (data freshness guarantee)
• ·FR3: Interactive SQL-like queries with GROUP BY, WHERE on any dimension, percentiles, count/sum/avg
• ·FR4: Priority sampling — 100% errors/exceptions, 100% p99+ latency events, configurable rate for normal events
• ·FR5: Time-series drill-down (1-min granularity last hour, 5-min last day, 1-hour last week)
• ·FR6: Dashboard infrastructure — saved queries, auto-refreshing, shareable, embeddable in incident war rooms

Non-Functional Requirements

Organizational Context

• ·Existing system: Scuba already exists — this is evolution/re-architecture, not greenfield
• ·Migration: Strangler fig pattern from current Scuba to next-gen system
• ·Meta stack: Kafka, Thrift, Scribe, MySQL, Memcached, Hive/HDFS, Spark, Presto
• ·Users: 1000+ internal teams. Culture: move fast, data-driven debugging during SEVs, <1s query expectation
• ·Timeline: Phase 1 in 3 months with Ingestion + Storage + Query teams (~12 engineers)

Out of Scope (Explicitly)

• ·Batch analytics (Hive/Spark pipelines — separate team)
• ·Time-series monitoring metrics (ODS/Gorilla — different access pattern)
• ·Full-text log search (LogDevice — different storage model)
• ·ML training pipelines (FBLearner — different latency/throughput tradeoffs)

Scale Numbers

Storage Estimation

Cost Estimation

Architectural implication: Memory fleet = 72% of total cost. Sampling strategy directly impacts total cost. Reducing average sample rate from 10% to 5% saves ~$1.5M/month. Priority sampling is the critical cost lever — we need per-table sampling configuration to optimize cost vs. fidelity tradeoffs.

External API (Engineer-Facing)

Inter-Service Contracts (Thrift — Meta convention)

API Governance

• ·Schema registry: Thrift IDL for all inter-service contracts
• ·Evolution: Backward-compatible only — additive fields, no removal or type changes
• ·Rate limiting: Per-user: 100 queries/min, per-dashboard: 60 refreshes/min
• ·Query cost estimation: Dry-run mode returns estimated partitions, rows, and latency before execution

Core Entities

Data Ownership Boundaries

Storage Technology Choices

Data Lifecycle

Architecture Diagram

Service Boundaries with Ownership

Sync vs Async Communication

Failure Domain Boundaries

Deep Dive 1: In-Memory Columnar vs Row Store

Recommendation: Pure in-memory columnar for hot tier. The 10-20x compression advantage and sub-second scan performance make it the only viable option for interactive analytics at this scale. Hybrid is a Phase 3 optimization.

In-Memory Columnar Storage (detailed):

Deep Dive 2: Sampling Strategy

Recommendation: Priority sampling with head-based trace preservation. Priority rules ensure 100% coverage of the events that matter most (errors, slow requests), while random sampling of normal events keeps costs manageable. Head-based trace preservation means if any event in a trace is kept, all events in that trace are kept — critical for debugging request flows.

Sampling decision flow (at ingester):

Deep Dive 3: Query Execution

Recommendation: Scatter-gather as primary execution model, with optional pre-aggregation for popular dashboards. Scatter-gather preserves full ad-hoc flexibility (Scuba’s key value proposition). Pre-aggregation is a Phase 2 optimization for dashboards that refresh every 30s and always run the same query.

Query execution engine:

Deep Dive 4: Schema Evolution

Recommendation: Backward-compatible evolution (additive changes only). This balances flexibility with data quality. New columns can be added freely; old events have NULL for new columns. Column type changes require a new column name (e.g., latency_ms INT → latency_us INT).

Schema evolution protocol:

Deep Dive 5: Multi-Datacenter Deployment

Recommendation: Each DC stores local events, cross-DC fan-out for global queries. This is the best cost/latency tradeoff. During incidents, engineers are typically debugging a specific DC’s services — local queries are fast (< 1s). For global aggregations (e.g., “total error rate worldwide”), the query coordinator fans out to all 3 DCs.

Multi-DC architecture:

Migration Strategy: Scuba v1 → v2 (Strangler Fig)

Scuba v1 exists and is used by 1000+ teams. Migration is the highest-risk aspect. Strangler fig: build v2 alongside, migrate table by table.

Phase 1: Shadow Mode (Weeks 1-8) — 3 high-value tables (feed_ranking, web_requests, errors), single DC, dual-ingest into old + new, compare query results within 1%. Teams: Ingestion + Storage + Query Engine.

Phase 2: Dashboard Migration (Weeks 9-16) — Query UI + dashboard builder, migrate top 50 dashboards, dogfood with Observability team, second DC (US-West). Teams: + Dashboard & Viz.

Phase 3: Broad Migration (Weeks 17-28) — Self-serve table creation portal, migrate 100 tables/week via automated tool, third DC (EU), per-table sampling config. Teams: + Schema & Metadata.

Phase 4: Decommission v1 (Months 8-12) — Old Scuba fully decommissioned, anomaly detection online, incident integration.

Rollback Plan

Bottleneck Analysis

Observability (RED Method)

Critical Alerts

1. Data freshness > 2 min for any table — ingestion pipeline stalled

2. Query latency p99 > 10s — query engine overloaded or hot partition

3. Memory utilization > 85% on any storage server — risk of OOM, trigger eviction

4. Kafka consumer lag > 5 min — ingesters falling behind, potential data loss window

5. Cross-DC replication lag > 10 min — global queries returning stale data

Distributed tracing: Event emit → Kafka publish (5ms) → broker replicate (10ms) → ingester consume+validate+sample (~15s avg) → columnar write → queryable. Key metric: end-to-end freshness. Target: p50 < 15s, p99 < 30s.

2-3 Year Evolution

Year 1: Foundation

• ·All 10K event tables migrated from Scuba v1
• ·3 DCs (US-East, US-West, EU) operational
• ·50,000+ active dashboards
• ·Query performance: p99 < 5s, data freshness < 30s
• ·Operational maturity: on-call runbook, automated rollback, chaos testing

Year 2: Intelligence

• ·Automated anomaly detection on event streams (statistical + ML-based)
• ·Correlation engine: link metric anomalies to specific event dimension changes
• ·Warm tier: on-disk columnar storage for 7-30 day range (extend retention without memory cost)
• ·Adaptive sampling: automatically adjust sample rates based on query patterns and event importance
• ·Dashboard templating: standard templates for common service types (web server, ML model, etc.)

Year 3: Platform

• ·ML root cause analysis: given an anomaly, automatically identify the dimensional combination explaining it (e.g., "error rate spike is 95% explained by country=BR AND app_version=7.2")
• ·Predictive alerting: detect trends before they become incidents
• ·Incident management integration: auto-open relevant dashboards in war room during SEVs
• ·Natural language query: "show me error rate by country for feed_ranking in the last hour"
• ·Self-serve derived metrics: define computed columns and materialized aggregations via config

Cost Optimization Roadmap

Key E6-level insights in this design:

1. In-memory columnar storage is the core architectural choice enabling sub-second aggregation queries over billions of rows — 10-20x compression via dictionary encoding and bit-packing makes this economically viable

2. Priority sampling is the most important cost lever — it determines whether the system costs $4M/month or $25M/month, while preserving 100% fidelity for the events that matter most (errors, tail latency)

3. Scatter-gather query execution with partition pruning ensures query fan-out is proportional to time range, not cluster size — a 1-hour query hits 15 servers, not 1000

4. Schema evolution without downtime via backward-compatible additive changes, with mixed-version partition handling in the query engine

5. Multi-DC with local-first queries optimizes for the common case (debugging your DC’s services during an incident) while supporting global aggregation via cross-DC fan-out

6. Strangler fig migration from existing Scuba v1, with dual-ingest validation and table-by-table migration — the highest-risk aspect of the project and the one that most requires staff-level coordination across 6 teams