Requirements and Scale
Functional: upload photos, view photos, follow users, generate a news feed of photos from followed users. Non-functional: highly available, latency < 200ms for feed, eventual consistency is acceptable.
Scale: 1 billion users, 500M daily active, 2M new photos/day (~23 photos/sec). Average photo size 200 KB → 2M × 200KB = 400 GB/day of new photo storage. Over 10 years: ~1.4 PB. Read-heavy workload.
High-Level Architecture
Three tiers: (1) Client apps → (2) Application servers behind a load balancer → (3) Storage layer (object store for photos, metadata DB for photo records, graph DB or social graph table for follows). Photos are never stored in the database — only metadata (photo_id, user_id, timestamp, location, caption). The actual image bytes go to an object store (S3) and are served through a CDN for fast global delivery.
graph TD Client --> LB["Load Balancer"] LB --> AppServer["App Servers"] AppServer --> PhotoMetaDB[(Photo Metadata DB MySQL sharded)] AppServer --> UserDB[(User DB MySQL)] AppServer --> S3["Object Store (S3)"] AppServer --> Cache["Redis Cache"] S3 --> CDN["CDN (CloudFront)"] CDN --> Client
Database Schema and Sharding
Photo table: photo_id (PK), user_id, photo_path (S3 URL), photo_latitude, photo_longitude, created_at. User table: user_id (PK), name, email, date_of_birth. UserFollow table: follower_user_id, followee_user_id.
Shard the Photo table by user_id so all photos for a user land on one shard. Shard key = user_id mod N. Problem: hot shard for celebrities. Solution: use consistent hashing and spread celebrity data across virtual nodes. Keep User and UserFollow tables on a different cluster since they are smaller and read-heavy.
Use two types of IDs: (1) auto-increment within a shard (fast), (2) global unique photo_id across shards needed for sorting feed by recency — solve with a sequence service (Twitter Snowflake-style) that embeds timestamp + shard ID + sequence into a 64-bit integer.
News Feed Generation
Three approaches: (1) Pull (fanout-on-read) — on page load, fetch all followees, query each shard for their latest photos, merge-sort, return top N. Very slow for users following thousands of people. (2) Push (fanout-on-write) — when a user posts, immediately push the photo_id into every follower's feed table. Feeds are pre-generated. Fast reads but wasted writes if followers are inactive; celebrities with millions of followers cause write storms. (3) Hybrid — for normal users, use fanout-on-write. For celebrities (>X followers), skip the push; instead, on feed read, fetch celebrity posts separately and merge with the pre-built feed. This is Instagram's actual approach.
graph LR Upload["User uploads photo"] --> FeedWorker["Feed Worker"] FeedWorker --> FollowerList["Fetch Follower List"] FollowerList --> FeedDB[(User Feed Tables Redis or Cassandra)] FeedDB --> FeedRead["Feed Read API"] FeedRead --> Client
Photo Upload and CDN
Upload flow: client sends photo to app server → app server gets a pre-signed S3 URL → client uploads directly to S3 (bypass app server for large blobs) → app server saves metadata to DB. CDN (like CloudFront) sits in front of S3. The CDN caches the photo at edge nodes near the user. Photo path stored in DB is the CDN URL, not the raw S3 URL. For thumbnails, a resizing service processes each uploaded photo and creates 3-4 size variants stored in S3.
Reliability and Replication
Each shard has a primary + 2 replicas. Writes go to primary, reads go to replicas. Photos in S3 are stored with 11 nines of durability (3 copies across 3 AZs). Metadata DB uses asynchronous replication with automatic failover via orchestrators like MHA. Cache (Redis) uses a cluster with sentinel for HA. Application servers are stateless — any server can handle any request.