Sharding

Backend Fundamentals · System Design · 9 min read

1. What problem does this solve?

Motivation
When a single database instance cannot keep up with the volume of data or traffic, you need a way to split the workload. Sharding is the technique of partitioning a dataset across multiple machines so that each shard handles a subset of the data and its queries.

What problem exists

• Capacity limits: A single PostgreSQL or MongoDB instance can only hold a few terabytes before performance degrades.
• Throughput limits: Even if the data fits, a single node can only process a limited number of concurrent requests.
• Availability: A single point of failure means the whole service goes down if that node crashes.

Why the problem matters

• Users experience slow queries or timeouts.
• New features cannot be released because the database cannot scale.
• SLA violations lead to revenue loss and reputational damage.

What happens if sharding is not used

• The database becomes a bottleneck.
• Scaling horizontally is impossible; you must buy a bigger machine (vertical scaling) which has limits.
• You risk data loss or corruption if the single node fails.

Where this concept appears in real backend systems

• Social media platforms partition user data by user ID.
• E‑commerce sites shard order histories by customer ID.
• Messaging services shard chat rooms by room ID.
• SaaS products often shard tenant data to isolate customers.

2. Mental model

Analogy
Think of a library that has only one shelf. As books (rows) pile up, the shelf becomes crowded and it takes longer to find a book. Sharding is like moving the library into multiple rooms, each with its own shelf. A book’s location is determined by its title (shard key). When you need a book, you go straight to the room that holds it, speeding up the search.

Engineering model

• Shard key: A column or set of columns that deterministically maps a row to a shard.
• Shard: A separate database instance (or logical partition) that stores a subset of rows.
• Routing layer: A component that, given a query, determines which shard(s) to hit.
• Rebalancing: Moving data between shards when the distribution changes.

3. Deep explanation

Internals

• Hash sharding: Compute hash(key) % N to pick a shard.
• Range sharding: Partition by key ranges (e.g., user_id 1–1M → shard 1).
• Directory sharding: Maintain a lookup table that maps keys to shards.
• Consistent hashing: Minimizes data movement when adding/removing shards.

Important terminology

• Shard key – the attribute used to decide shard placement.
• Shard map – metadata that maps keys to shard addresses.
• Hot shard – a shard that receives disproportionate traffic.
• Rebalancing – redistributing data to even out load.
• Cross‑shard query – a query that touches multiple shards (often expensive).

Trade‑offs

Failure modes

• Shard loss: If a node dies, its data is unavailable. Requires replication or backup.
• Routing failure: If the shard map is stale, queries may go to wrong nodes.
• Rebalancing race: Concurrent writes during data movement can be lost or duplicated.

Limitations

• Cross‑shard joins are expensive or impossible without a distributed query engine.
• Atomicity across shards is hard; you need distributed transactions or eventual consistency.
• Schema evolution must be applied to all shards.

Where this fits in a backend architecture

┌─────────────────────┐
│  Client / API Layer │
└───────┬──────────────┘
        │
┌───────▼──────────────┐
│  Routing / Shard Map │
└───────┬──────────────┘
        │
┌───────▼──────────────┐
│  Shard 1 (Postgres)  │
│  Shard 2 (Postgres)  │
│  Shard 3 (Mongo)     │
└──────────────────────┘

The routing layer is often implemented as a thin service or middleware that consults a shard map (stored in Redis or a config file).

Interaction with related concepts

• Replication: Each shard can be replicated for HA.
• Caching: Redis can cache shard map and hot query results.
• Load balancer: Distributes API traffic to routing layer instances.
• Observability: Metrics per shard (latency, QPS) help detect hot shards.

4. How it works

1. Choose a shard key (e.g., user_id).
2. Determine shard algorithm (hash, range, directory).
3. Store shard map (e.g., in Redis: user_id → shard_url).
4. Incoming request → routing layer looks up shard.
5. Routing layer forwards the query to the target shard.
6. Shard processes the query and returns results.
7. Routing layer aggregates (if needed) and returns to client.

sequenceDiagram
    participant Client
    participant Router
    participant Shard1
    participant Shard2

    Client->>Router: GET /users/12345
    Router->>Router: lookup shard for 12345
    Router->>Shard1: SELECT * FROM users WHERE id=12345
    Shard1-->>Router: {id:12345, name:"Alice"}
    Router-->>Client: {id:12345, name:"Alice"}

5. Practical implementation

Example: Hash sharding with PostgreSQL and Node.js

// shardRouter.ts
import { Pool } from 'pg';
import crypto from 'crypto';

const shards = [
  { id: 0, url: 'postgres://user:pass@shard0.db:5432/app' },
  { id: 1, url: 'postgres://user:pass@shard1.db:5432/app' },
  { id: 2, url: 'postgres://user:pass@shard2.db:5432/app' },
];

const pools = shards.map(s => new Pool({ connectionString: s.url }));

/**
 * Simple hash function: returns shard index for a given key.
 */
function getShardIndex(key: string | number): number {
  const hash = crypto.createHash('md5').update(String(key)).digest('hex');
  // Convert first 8 hex chars to integer
  const intHash = parseInt(hash.slice(0, 8), 16);
  return intHash % shards.length;
}

/**
 * Execute a query on the correct shard.
 */
export async function queryShard(key: string | number, sql: string, params: any[]) {
  const idx = getShardIndex(key);
  const pool = pools[idx];
  return pool.query(sql, params);
}

// userService.ts
import { queryShard } from './shardRouter';

export async function getUser(userId: number) {
  const res = await queryShard(
    userId,
    'SELECT id, name, email FROM users WHERE id = $1',
    [userId]
  );
  return res.rows[0];
}

Key points

• getShardIndex deterministically maps a user ID to a shard.
• All writes and reads for a user go to the same shard, ensuring consistency.
• Adding a new shard requires updating the shards array and rebalancing data.

Range sharding example (simplified)

// rangeRouter.ts
const shardRanges = [
  { id: 0, min: 1, max: 1000000, url: 'postgres://...' },
  { id: 1, min: 1000001, max: 2000000, url: 'postgres://...' },
];

function getRangeShard(id: number) {
  return shardRanges.find(r => id >= r.min && id <= r.max)!;
}

6. Production considerations

7. Common mistakes

8. Senior engineer perspective

What changes at scale

• Routing layer becomes a distributed service (e.g., multiple router pods behind a load balancer).
• Shard map stored in a highly available store (Redis cluster, etcd).
• Monitoring dashboards show per‑shard health; alerts for hot shards.
• Automated rebalancing pipelines run during low‑traffic windows.

Design decisions

• Consistent hashing vs. range: Consistent hashing reduces data movement but can lead to uneven distribution if key space is sparse.
• Replication factor: 3 replicas per shard vs. 2 for cost vs. availability trade‑off.
• Cross‑shard transaction strategy: Use Saga pattern or eventual consistency.
• Cache invalidation: Decide whether to cache shard map in each router or use a shared Redis cache.
• Deployment topology: Prefer colocating shards with application instances to reduce latency.

Bottlenecks

• Router latency: Must be sub‑millisecond; otherwise it dominates query time.
• Network hops: Each shard query adds latency; consider colocating routers and shards.
• Rebalancing windows: Moving large ranges can lock tables; plan for downtime or use online migration tools.

Monitoring

• Shard latency: Histogram per shard.
• Error rates: 5xx per shard.
• Replica lag: For read replicas.
• Resource usage: CPU, memory, disk I/O per shard.

Cost/performance

• More shards → more nodes → higher cost but better throughput.
• Replication adds storage cost but improves availability.
• Sharding strategy can affect query cost (e.g., cross‑shard queries cost more).

When NOT to use sharding

• Small datasets (< 10 GB) where a single node suffices.
• Highly relational workloads that require many cross‑table joins.
• Strict ACID requirements across all data that cannot tolerate eventual consistency.

9. Interview questions

10. Hands‑on task

Task: Add a new shard to the existing hash‑sharded PostgreSQL setup in the repository.

1. Spin up a new PostgreSQL container (e.g., shard3).
2. Update shardRouter.ts to include the new shard.
3. Rebalance a subset of users (e.g., IDs 1 000 001–2 000 000) to the new shard by running a migration script.
4. Verify that reads/writes for those users now hit the new shard.
5. Update the Docker Compose file to expose the new shard and restart the stack.

Hints: Use the getShardIndex function to determine which IDs should go to the new shard. Use pg_dump/pg_restore or COPY for data migration.

11. Quick revision

• Sharding partitions data across multiple nodes to scale capacity and throughput.
• Shard key: deterministic mapping of rows to shards.
• Hash sharding: even distribution, hard to range‑query.
• Range sharding: efficient range queries, risk of hot shards.
• Directory sharding: lookup table for key → shard.
• Consistent hashing: minimal data movement on topology changes.
• Routing layer must be highly available and fast.
• Replication per shard is essential for HA.
• Cross‑shard joins are expensive; redesign schema if needed.
• Monitor per‑shard latency, error rates, and replica lag.
• Rebalancing requires careful migration and map updates.
• Avoid sharding for small, simple workloads.

12. References

• PostgreSQL documentation – Sharding
• MongoDB documentation – Sharding
• Consistent hashing – Wikipedia
• Hash ring implementation – HashRing
• Distributed tracing – OpenTelemetry docs – Tracing
• Docker Compose – Official docs
• Redis – Cluster