Database Sharding Strategies

Production engineering guide for database sharding strategies covering patterns, implementation strategies, and operational best practices.

Database sharding strategies are essential when a single database instance cannot handle the scale of data, query load, or transactional throughput. Sharding splits a large database into smaller, independent databases (shards), each managing a subset of the data, enabling horizontal scaling. This approach becomes critical when a database’s performance degrades under increasing load, when a single node’s storage capacity is exhausted, or when read and write latency exceeds acceptable thresholds. Sharding is the next step after vertical scaling and replication, and it is the primary solution for systems with high throughput and large datasets—such as social media feeds, real-time analytics, or global e-commerce platforms.

Choosing a Shard Key

The shard key determines how data is distributed across shards. The choice directly impacts query performance, data locality, and the ability to scale both reads and writes. The shard key must be a column or set of columns that are frequently used in queries, ideally present in the primary key of the table. It should have high cardinality, be relatively stable, and support efficient range and equality queries.

For example, in an e-commerce system, user_id is a natural shard key: most queries—user profiles, order history, cart contents—are filtered by user. A user_id of 12345 maps to a specific shard, and all data for that user resides on a single database node.

To define a shard key in a sharded environment, a metadata table is maintained that maps shard keys to shard identifiers. This is often implemented as a shard_map table:

CREATE TABLE shard_map (
    shard_key_start BIGINT,
    shard_key_end BIGINT,
    shard_id TEXT,
    host TEXT,
    port INTEGER,
    created_at TIMESTAMP
);

The system uses this table to route queries: a request for user_id = 12345 first queries shard_map to locate the shard containing user_id 12345, then routes the query to that shard’s database instance.

Shard key selection is not trivial. A poor choice leads to uneven data distribution, hot shards, and inefficient queries. For instance, using order_id as a shard key in a system where users create many orders but each user creates only a few orders leads to underutilized shards and scattered data. A common failure mode is the "shard key explosion": when the shard key is a composite of many columns, and the combination of values results in high cardinality but poor query locality.

Implementing Range-Based Sharding

Range-based sharding divides data into contiguous ranges of the shard key. Each shard is responsible for a range of values, such as user_id from 0 to 99,999, 100,000 to 199,999, and so on. This strategy excels for workloads with range queries (e.g., “get all orders from the last 30 days”) and supports efficient data distribution.

In a range-based setup, the shard_map table is populated with ranges:

INSERT INTO shard_map (shard_key_start, shard_key_end, shard_id, host, port)
VALUES
    (0, 99999, 'shard-01', 'db-01.example.com', 5432),
    (100000, 199999, 'shard-02', 'db-02.example.com', 5432),
    (200000, 299999, 'shard-03', 'db-03.example.com', 5432);

When a new user with user_id = 150000 is created, the system first checks if the range 100000–199999 is already assigned to shard-02. If so, the new user’s data is written to db-02.example.com:5432.

The primary challenge in range-based sharding is data skew—some ranges become much larger than others. A sudden spike in user sign-ups from a specific region can cause one shard to grow rapidly while others remain underutilized. This imbalance leads to hot shards and uneven load distribution.

To address this, systems implement range splitting. When a shard exceeds a threshold size (e.g., 100 GB), it is split into two smaller ranges. The split process involves:

  1. Creating a new shard entry in shard_map for the new range.
  2. Copying data from the old shard to the new one.
  3. Updating the shard_map with the new range mapping.
  4. Redirecting new writes to the new shard.
  5. Migrating existing data asynchronously.

The process must be coordinated to avoid downtime. A common failure mode occurs when the split is not atomic: writes to the original shard during the split are lost or duplicated, leading to data inconsistency.

Implementing Hash-Based Sharding

Hash-based sharding uses a hash function to map shard keys to shards. The most common method is consistent hashing, which minimizes re-sharding when the number of shards changes.

Given a set of shards, a hash function (e.g., crc32 or md5) maps each shard key to a bucket. The buckets are arranged in a ring, and each shard is responsible for a segment of the ring.

For example, with four shards (shard-01, shard-02, shard-03, shard-04) and a user_id as the key:

This approach ensures even data distribution and minimal re-sharding when adding or removing shards.

A practical implementation involves a shard_map table with shard_id, hash_start, hash_end, and target:

CREATE TABLE shard_map (
    shard_id TEXT,
    hash_start BIGINT,
    hash_end BIGINT,
    host TEXT,
    port INTEGER
);

To find the shard for a user_id, the system computes the hash, then queries shard_map to find the shard where hash_start ≤ hash(user_id) < hash_end.

Hash-based sharding performs well for point queries and insert-heavy workloads. However, it struggles with range queries: to retrieve all users from user_id = 10000 to 20000, the system must query multiple shards and merge results.

A common failure mode is hash collisions—multiple keys mapping to the same shard, leading to uneven data distribution. This is especially problematic when the hash function is not well-distributed or when the load profile changes over time.

Cross-Shard Queries and Joins

Sharding introduces complexity in queries that span multiple shards. A simple SELECT * FROM orders WHERE user_id = 12345 is efficient—only one shard needs to be queried. But a JOIN between orders and users is more complex.

To execute a cross-shard JOIN, the system must:

  1. Query each shard for the relevant user records.
  2. Gather the results into a single dataset.
  3. Perform the join on the client side or using a distributed query engine.

For example, to get all orders and user details for users in shard-01 and shard-02:

-- Execute on each shard
SELECT o.*, u.name, u.email
FROM orders o
JOIN users u ON o.user_id = u.id
WHERE o.shard_id IN ('shard-01', 'shard-02');

This query may be executed in parallel across multiple shards, with results collected and merged.

To improve performance, systems use distributed query planning. A coordinator node (or query router) parses the SQL, determines which shards are needed, and dispatches subqueries. The results are merged and returned to the client.

A frequent failure mode is inefficient query routing. If the shard key is not properly indexed on the user_id column, a cross-shard JOIN becomes slow. Another common issue is partial results: when one shard is unreachable, the query returns data from available shards, but the result set is incomplete. A client may fail to handle this by assuming all data is present.

Handling Shard Failures and Rebalancing

A shard can fail due to network issues, node crashes, or storage exhaustion. When a shard becomes unavailable, the system must:

  1. Detect the failure (via heartbeat or health checks).
  2. Mark the shard as down in shard_map.
  3. Redirect queries to a backup shard or a fallback set of shards.
  4. Rebalance data to restore availability.

For example, if shard-02 goes down, the system updates shard_map to mark it as status = 'down' and routes all user_id queries for shard-02 to shard-01 as a fallback.

Rebalancing involves redistributing data across shards to achieve load balance. This is triggered by metrics such as:

Rebalancing is typically done in two phases:

  1. Data migration: using rsync, pg_dump, or COPY to transfer data from source to target shards.
  2. Routing update: updating shard_map to reflect new assignments.

A critical failure mode is data inconsistency during migration. If the source shard is still accepting writes during migration, and the target shard is not yet up-to-date, the system may return stale or duplicated data. This is especially problematic for write-heavy workloads.

To ensure consistency, systems use dual-write or two-phase commit patterns:

Handling Global Data and Reference Tables

Sharding is most effective for application data—data that is naturally partitioned by the shard key. However, reference data—data that is shared across all shards—requires special handling.

Examples of reference data include:

These tables are typically replicated across all shards. A common strategy is read-replica replication:

For instance, a settings table is updated via:

UPDATE settings SET value = '2024' WHERE key = 'current_year';

This change is logged in a change_log table, and a background process (e.g., a pg_cron job) applies the changes to each shard:


psql -h db-01.example.com -p 5432 -U app -d app_db -c "
  INSERT INTO settings (key, value) 
  SELECT key, value 
  FROM change_log 
  WHERE applied_at IS NULL;
"

A failure mode is stale reference data: when a shard has not applied recent changes from the master. A user on shard-01 may see outdated settings while another on shard-02 sees the latest version.

To mitigate this, systems use reference table caching. A local cache (e.g., in memory or Redis) holds the most frequently accessed reference data. The cache is invalidated on updates, and queries first check the cache before hitting the database.

Managing the Shard Map

The shard_map table is the central source of truth for routing. It must be highly available and consistent. A failure in shard_map—such as a node crash or network partition—can cause the entire database to become unreachable.

To ensure availability, the shard_map is often stored in a distributed key-value store (e.g., Consul, ZooKeeper, etcd) or a replicated database. A typical configuration uses:

# etcd configuration
shard_map:
  /shards/01:
    host: db-01.example.com
    port: 5432
    range_start: 0
    range_end: 99999
  /shards/02:
    host: db-02.example.com
    port: 5432
    range_start: 100000
    range_end: 199999

A common failure mode is inconsistent shard map updates: when a new shard is added, but the shard_map is updated on one node but not others. This leads to routing errors and partial data access.

To prevent this, systems use atomic updates via transactions or versioned keys. Each shard_map entry has a version number, and updates are done using compare-and-swap:

# etcd: update shard-02 with version check
etcdctl put /shards/02 \
  --value '{"host":"db-02.example.com","port":5432,"range_start":100000,"range_end":199999}' \
  --with-prev-kv \
  --prev-value='{"host":"old-db.example.com","port":5432,"range_start":100000,"range_end":199999}'

If the previous value does not match, the update fails, ensuring consistency.

This page was rewritten on 17 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.