⚡ MongoDB Sharding

Answer:

A sharded cluster consists of three main components:

1. mongos (Query Router):

• Acts as interface between application and cluster
• Routes queries to appropriate shard(s)
• Merges results from multiple shards
• Stateless - can have multiple mongos instances
• Caches metadata for faster routing

2. Config Servers (Metadata Store):

• Store cluster metadata (chunk ranges, shard locations)
• Always deployed as a replica set (3+ servers)
• Critical - if down, cluster becomes read-only
• Contain: chunk mappings, zone configurations, balancer state

3. Shards (Data Storage):

• Each shard is a replica set storing subset of data
• Contains specific chunks based on shard key ranges
• Can scale by adding more shards
• Typically 2-3 members per shard for high availability

Data Flow: Application → mongos → (checks config servers) → routes to shard(s) → returns merged results

Answer:

Shard Key: The indexed field(s) MongoDB uses to partition data. It's IMMUTABLE after selection!

Characteristics of Good Shard Keys:

• High Cardinality: Many unique values (userId, email) not few (gender, boolean)
• Even Distribution: Values spread evenly, not clustered (hashed keys help)
• Query Isolation: Most queries should target single shard (include in WHERE)
• Not Monotonic: Avoid always-increasing values (_id, timestamp) = all writes to one shard
• Frequently Queried: Should appear in most query filters

Examples:

• ✅ Good: { userId: 1 }, { email: "hashed" }, { region: 1, userId: 1 }
• ❌ Bad: { _id: 1 }, { createdAt: 1 }, { status: 1 }

Selection Process:

1. Analyze query patterns (what fields in WHERE clauses?)
2. Check cardinality (how many unique values?)
3. Test distribution (will data spread evenly?)
4. Consider growth (how will writes distribute over time?)
5. Prototype with production-like data

Answer:

Ranged Sharding:

• How: Divides data based on ranges of shard key values
• Distribution: Documents with nearby values on same shard
• Pros: Excellent for range queries, sorted access, related data together
• Cons: Risk of hot spots, uneven distribution with monotonic keys
• Use when: Range queries common, need data locality
• Example: userId 0-33M on shard1, 33M-66M on shard2, etc.

Hashed Sharding:

• How: Uses hash function on shard key for distribution
• Distribution: Guarantees even distribution regardless of key pattern
• Pros: Perfect balance, no hot spots, works with monotonic keys
• Cons: Range queries = broadcast to all shards, can't sort by shard key
• Use when: Even distribution critical, point queries common
• Example: hash(userId) distributes randomly across all shards

Choice Depends On:

• Query patterns: Range queries → Ranged, Point queries → Either
• Key characteristics: Monotonic → Hashed, Well-distributed → Ranged
• Hot spots: Concern → Hashed, Not concern → Ranged

Answer:

Chunks:

A chunk is a contiguous range of shard key values. MongoDB doesn't distribute individual documents - it distributes chunks.

• Default size: 128MB (configurable 1MB-1024MB)
• Indivisible: A chunk lives entirely on one shard
• Metadata: Chunk ranges stored in config servers

Chunk Splitting:

• Trigger: When chunk exceeds size limit (128MB)
• Process: MongoDB automatically splits into two smaller chunks
• Example: Chunk 0-50M (150MB) → splits to 0-25M (75MB) + 25M-50M (75MB)
• Location: Both new chunks remain on same shard initially

Chunk Migration:

• Purpose: Maintain even distribution across shards
• Trigger: Balancer detects imbalance (shard has too many/few chunks)
• Process: Balancer moves chunks from overloaded to underloaded shards
• Timing: Runs in background during configurable window
• Impact: Minimal - happens incrementally, doesn't block operations

Example Flow:

1. Chunk grows to 150MB → splits into two 75MB chunks
2. Shard1 now has 100 chunks, Shard2 has 50
3. Balancer migrates 25 chunks from Shard1 to Shard2
4. Result: Both shards have ~75 chunks (balanced)

Answer:

The Balancer: A background process running on the primary of the config server replica set that maintains even chunk distribution across shards.

How It Works:

1. Monitors: Continuously checks chunk distribution across shards
2. Detects Imbalance: Compares chunk counts per shard
3. Calculates Migrations: Determines which chunks to move where
4. Migrates: Moves chunks from overloaded to underloaded shards
5. Updates Metadata: Updates config servers with new chunk locations

Key Characteristics:

• Automatic: Runs without manual intervention
• Throttled: Limits concurrent migrations to avoid overload
• Schedulable: Can configure active windows (e.g., off-peak hours)
• Pauseable: Can stop/start for maintenance

Configuration:

sh.getBalancerState()      // Check if running
sh.stopBalancer()          // Stop balancer
sh.startBalancer()         // Start balancer
sh.setBalancerState(true)  // Enable balancer

When to Disable:

• During backups (to ensure consistent snapshots)
• Large bulk data loads (re-enable after)
• Maintenance windows on shards
• Critical high-traffic periods

Answer:

Targeted Query:

• Definition: Query includes shard key in WHERE clause
• Routing: mongos routes to specific shard(s) only
• Performance: Very fast - only queries necessary shards
• Example: db.users.find({ userId: 12345 }) → Goes to ONE shard
• explain() shows: Single shard accessed

Broadcast Query:

• Definition: Query does NOT include shard key
• Routing: mongos sends query to ALL shards
• Performance: Slow - must query every shard and merge results
• Example: db.users.find({ email: "..." }) → Goes to ALL shards
• explain() shows: SHARD_MERGE, all shards queried

Performance Impact:

Best Practices:

• Always include shard key when possible
• Use compound indexes for non-shard-key queries
• Monitor broadcast query frequency
• Consider secondary indexes for common non-shard-key queries

Answer:

Key Monitoring Commands:

sh.status()                        // Overall cluster status
db.collection.getShardDistribution() // Per-collection distribution
sh.getBalancerState()              // Balancer status
db.adminCommand({listShards: 1})   // List all shards

Critical Metrics:

• Chunk Distribution: Should be roughly equal (±20%) across shards
• Balancer Activity: Check migration frequency and failures
• Query Performance: Use explain() to verify targeted queries
• Shard Resources: CPU, RAM, disk I/O per shard
• Connection Pools: mongos to config servers and shards

Common Issues & Solutions:

• Uneven Distribution:
  • Check shard key choice (might be poor)
  • Verify balancer is running
  • Consider pre-splitting chunks
• Slow Queries:
  • Check if broadcast queries (add shard key)
  • Verify indexes on each shard
  • Use explain() to diagnose
• Hot Shard:
  • Monotonic shard key causing all writes to one shard
  • Solution: Switch to hashed sharding or compound key
• Balancer Not Running:
  • Check if manually disabled
  • Verify config server primary election
  • Check for failed migrations in logs

Monitoring Tools:

• MongoDB Atlas (built-in monitoring)
• MongoDB Ops Manager / Cloud Manager
• Third-party: Datadog, New Relic, Prometheus
• Custom scripts using db.serverStatus()

// Create zones
sh.addShardToZone("shard-us", "USA")
sh.addShardToZone("shard-eu", "EUROPE")
sh.addShardToZone("shard-in", "INDIA")

// Assign ranges
sh.updateZoneKeyRange("mydb.users",
  { country: "US" }, { country: "US" }, "USA")
sh.updateZoneKeyRange("mydb.users",
  { country: "GB" }, { country: "GB" }, "EUROPE")