System design deep dive · HLD
Design a Social Graph & "People You May Know": full high-level design.
Requirements, back-of-envelope capacity estimation for billions of users and hundreds of billions of follow edges, a high-level design diagram, a deployment architecture diagram, sequence diagrams for the follow-edge write path and the offline recommendation pipeline, and an entity-relationship diagram - with the reasoning an interviewer expects behind every box and arrow.
1. Clarify requirements before drawing any box
A social graph looks simple - "who follows whom" - until scale forces every read and write to be reasoned about separately. The clarifying step is deciding which model is primary and which questions must never touch a live, unbounded graph traversal.
Functional requirements
Non-functional requirements
2. Back-of-the-envelope capacity estimation
The two numbers that drive every decision below are the total edge count (it decides partitioning) and the read:write ratio (it decides how aggressively hot nodes must be cached).
| Metric | Assumption | Resulting estimate |
|---|---|---|
| Total users | 2 billion | Base population for the graph |
| Average out-degree | ~300 average following count, heavily skewed - celebrity accounts have tens of millions of followers | 2B × 300 ≈ 600 billion directed edges |
| Storage per edge | follower_id + followee_id (8B each) + created_at (8B) + edge_type/flags (~1B, padded) ≈ 40 bytes | 600B × 40B ≈ 24TB raw |
| Total graph storage | Both directions stored (followers-of + following-of) × 3-way replication | 24TB ×2 ×3 ≈ ~144TB across the cluster |
| Edge write rate | ~500 million follow/unfollow actions per day globally | ≈ 5,800 writes/sec average, ≈ 20,000/sec peak (viral moments) |
| Edge read rate | Profile views + feed generation checking follow-status or fetching following-lists | ≈ 2,000,000 reads/sec average - roughly 300:1 read:write |
| Offline PYMK batch cost | 2B users × capped 200 sampled 1-hop neighbors × capped 100 sampled 2nd-hop neighbors each | ≈ 40 trillion bounded edge-touches/night, vs an uncapped traversal from one 10M-follower hub alone approaching 10^14 |
3. High-level design (HLD)
The HLD separates the strongly-consistent direct-edge path from the eventually-consistent recommendation path - they have different latency budgets, different consistency requirements, and almost never share a code path at request time.
What each box owns
Graph API (create/remove edge, get-edges)
The single stateless entry point for all edge operations. Writes are routed through the follow write path; reads are routed through the routing layer to the owning shard, checking the edge cache first. It never performs a live 2-hop traversal itself - PYMK requests are simply a read against the precomputed recommendation candidate store.
Graph partitioning / routing layer
Maps a user_id to the shard that owns that user's adjacency lists using consistent hashing, so adding or removing shards only reshuffles a small fraction of users instead of the whole ring. Because the routing key is always a known user_id, every "get all edges for user X" request resolves to exactly one shard - no scatter-gather across the cluster.
Follow write path (dual-write, strongly consistent)
A single follow action touches two adjacency lists: the follower's "following" list and the followee's "followers" list. Both writes are executed as one logically atomic operation (retried together on partial failure) so the UI can never observe a state where A appears to follow B but B's followers list doesn't yet contain A.
Sharded edge store and edge cache
The sharded edge store (a wide-column store like Cassandra, or a TAO-style graph cache layered over sharded MySQL) holds the durable adjacency lists, partitioned by owning user_id. A dedicated Redis-based edge cache sits in front of it purely to absorb the read load from celebrity/hub accounts, whose follower lists would otherwise dominate a single shard's traffic.
PYMK batch pipeline and recommendation candidate store
A nightly Spark job reads a snapshot of the edge store, performs a capped, sampled 2-hop traversal per user, scores candidates by mutual-connection count, and writes the top-N results into the recommendation candidate store. The online path never traverses the graph live - it only ever reads whatever the last batch run precomputed.
4. Detailed architecture diagram
The architecture diagram commits to specific technology and deployment shape: how many shards, how replicas are laid out, and - critically for this system - where a dedicated hot-node tier sits to absorb the wildly uneven traffic that celebrity accounts generate.
| Decision | Choice | Reasoning |
|---|---|---|
| Partitioning strategy | Hash-based user_id sharding via consistent hashing | Even distribution and simple resharding; friends often land on different shards, but "list my edges" never needs shard-mates to be co-located - only single-shard, single-key lookups matter. |
| Storing both edge directions | Followers-of and following-of both persisted (redundant) | Doubles write cost and storage, but keeps every read O(1) on a single shard; deriving one direction from the other at read time would require a cross-shard scan. |
| PYMK computation | Offline, precomputed, sampled 2-hop batch job | A hub node's full 2-hop expansion is combinatorially infeasible online; bounding fan-out per hop and running nightly trades exhaustiveness for a request-time cost of a single cache read. |
| Partitioning alternative | Community-aware / edge-cut-minimizing partitioning | Better locality for genuinely social access patterns, but far harder to maintain as the graph mutates continuously - not worth it when direct-edge lookups are already single-shard by construction. |
5. Sequence diagrams for the two critical flows
The first flow is on the strongly-consistent critical path (a user must never see a stale follow state); the second is entirely offline except for the final, trivially fast read.
5.1 Create a follow edge, then read the following list
Steps 4-5 are drawn as a single logical unit deliberately: both adjacency-list writes must succeed together, and on a partial failure the operation is retried rather than leaving the graph in a state where A follows B but B's followers list doesn't yet contain A. Step 6 is fire-and-forget - cache invalidation is an optimization for the next read, not a correctness requirement for the write itself. On the read side, a cache hit (step 10) collapses the whole lookup to one round trip; the miss path noted below the diagram falls back to the sharded store and repopulates the cache before returning, which is why celebrity nodes need that cache tier to almost never miss.
5.2 Offline PYMK batch computation, then online serving
Steps 3-5 are the entire reason this system stays feasible at scale: capping fan-out to a fixed sample size at each hop turns an unbounded, per-node-dependent traversal cost into a fixed, predictable cost per user, independent of whether that user (or their neighbors) happen to be a hub with tens of millions of edges. Steps 8-11 are deliberately trivial - the online path never re-runs any of steps 3-5, it only ever reads whatever the last nightly run already computed, which is what keeps PYMK requests as fast as any other cache read.
6. Entity-relationship (ER) diagram and schema
The schema has to answer: what's the single key that makes "get all my edges" a one-shard lookup, how is a directed edge represented so both directions are O(1) to query, and how are precomputed recommendations refreshed without becoming their own bottleneck.
Key modeling decisions
| Storage choice | Use when | Watch out for |
|---|---|---|
| Wide-column, sharded by user_id (Cassandra) | Access pattern is overwhelmingly "get all edges for user X", at hundreds of billions of rows. | Ad-hoc queries like "who follows both A and B" need external indexing - the store is optimized for single-key access, not joins. |
| TAO-style graph cache over sharded MySQL | You want a proven relational engine underneath plus a purpose-built read-through cache that already understands edge semantics. | More moving parts to operate than a single wide-column cluster; cache-consistency logic is bespoke, not free. |
7. Deep dives interviewers actually probe
Why is naive live 2-hop friend-of-friend traversal infeasible at scale?
A full 2-hop expansion from a node with 10 million followers means visiting 10 million 1-hop neighbors and, for each of them, however many neighbors they have - a combinatorial fan-out that can reach hundreds of billions of pairs from a single request. No amount of horizontal scaling makes an unbounded, per-node-dependent cost acceptable on a request that must return in milliseconds. Capping the sample size at each hop turns the cost into a fixed constant per user, at the cost of not being exhaustive - an acceptable trade for a "may know" suggestion, which was never meant to be complete.
// bounded, sampled 2-hop traversal - cost is O(HOP1_CAP * HOP2_CAP), not O(degree^2)
const HOP1_CAP = 200, HOP2_CAP = 100;
function samplePymk(userId) {
const hop1 = sampleNeighbors(userId, HOP1_CAP); // capped, not full adjacency list
const scores = new Map();
for (const n of hop1) {
const hop2 = sampleNeighbors(n, HOP2_CAP); // capped again, even if n is a hub
for (const candidate of hop2) {
if (candidate === userId || alreadyConnected(userId, candidate)) continue;
scores.set(candidate, (scores.get(candidate) || 0) + 1); // mutual-connection count
}
}
return topN(scores, 50);
}How do you partition hundreds of billions of edges so "get all my edges" is one shard?
Hash the owning user_id through a consistent-hashing ring to pick a shard, and store both directions of every relationship (a followers-of row on the followee's shard, a following-of row on the follower's shard). This deliberately does not try to co-locate friends' data on the same shard - that would require constant, expensive rebalancing as the social graph evolves. The trade is redundant storage and double writes, in exchange for every read being a single-shard, single-key lookup with no fan-out.
How do you handle celebrity accounts read constantly and mutated frequently?
A dedicated hot-node cache tier (Redis) sits in front of the sharded store specifically so a small number of accounts don't overwhelm one shard's IOPS; a hub-node detector flags accounts crossing a follower-count threshold so their entries get pinned with longer TTLs. Follower and following counts shown in the UI are denormalized counters on the users table, incremented asynchronously off the write path - counting live from the edges table on every profile view would turn a cheap read into a full partition scan for exactly the accounts that get the most traffic.
What consistency does "did I follow this person" need versus "how many mutual friends"?
The direct edge itself must be strongly consistent: the dual write (following-of + followers-of) is retried as one unit, and a read immediately after a successful write must reflect it - a user unfollowing someone and then seeing "Follow" still greyed out is a trust-breaking bug. Mutual-friend counts, follower totals, and PYMK scores are all derived, aggregate figures computed from a snapshot that may be up to a day old; eventual consistency there is not just acceptable, it's the entire premise that makes the offline batch pipeline affordable.
How would this change for a symmetric mutual-friend model (Facebook) instead of follow?
A symmetric model needs a pending state: a friend request creates one edge row with status=pending, and only accepting it promotes it to status=accepted and materializes both directions (A→B and B→A) so lookups from either side stay O(1) - the storage shape barely changes, only the state machine in front of the write path does. Partitioning is unaffected: the edge is still keyed by the pair of user ids and sharded the same way, since "friend" is just an edge_type variant of the same edges table rather than a different schema.
Post a Comment
Add