How to Estimate Sharding Capacity: Calculating Required Databases and Tables for an Alibaba Interview

Why Sharding?

When a database reaches read or write bottlenecks that cannot be solved by SQL or index tuning, active connections approach the database’s limit, causing service crashes. The root causes are classified as disk I/O, network I/O, and CPU bottlenecks. Sharding (splitting databases and tables) is the remedy when performance problems stem from resource limits.

Two Design Principles

Hardware‑bound bottlenecks – if the issue is insufficient hardware resources, add more data sources (master‑slave clusters).

Table‑level bottlenecks – if a single table is too large or lock‑contentious, apply table‑level sharding.

Capacity Estimation

Estimation relies on two factors: existing data volume (only the online, query‑heavy data matters) and growth trend (first‑order and second‑order derivatives of data growth). Companies often use a three‑year plan or product‑manager KPIs to project growth; for example, a 100% yearly increase is assumed when the business is expected to double.

Example "big‑budget" estimate: start with 32 databases, each containing 32 tables (total 1024 tables). Assuming each database can handle 1,000 writes per second, the cluster supports 32 × 1,000 = 32,000 writes/s; with 1,500 writes per database, it reaches 48,000 writes/s. If each table stores 5 million rows, the whole cluster can hold 5 billion rows, which is sufficient for most large‑scale internet companies.

Sharding Schemes

1. Range (Continuous) Sharding

Data is split by a field range (e.g., user ID). Adding a new node only requires assigning a new range, no data migration.

class RangeSharding {
    static final long NODE0_MAX = 1000_0000L;
    static final long NODE1_MAX = 2000_0000L;
    int shard(long userId) {
        if (userId <= NODE0_MAX) return 0;
        if (userId <= NODE1_MAX) return 1;
        return 2;
    }
}

2. Modulo Sharding

The shard key is taken modulo the number of nodes. Simple and fast, but expanding the node count forces massive data migration and makes range queries inefficient.

class ModSharding {
    final int nodeCount;
    ModSharding(int nodeCount) { this.nodeCount = nodeCount; }
    int shard(long id) { return (int) (id % nodeCount); }
}

3. Consistent‑Hash Sharding

Uses a hash ring with virtual nodes; adding or removing a physical node only moves a small fraction of data.

import java.util.*;
class ConsistentHashSharding {
    final int VIRTUAL = 200;
    final SortedMap<Integer, Integer> ring = new TreeMap<>();
    final List<Integer> nodes = Arrays.asList(0,1,2);
    ConsistentHashSharding() {
        nodes.forEach(node -> {
            for (int i = 0; i < VIRTUAL; i++) {
                ring.put(hash(node + "#" + i), node);
            }
        });
    }
    int shard(String key) {
        int h = hash(key);
        if (!ring.containsKey(h)) h = ring.ceilingKey(h);
        return ring.get(h);
    }
    int hash(String s) { return s.hashCode() & 0x7fffffff; }
}

4. Snowflake‑Based Sharding

Snowflake generates globally unique, time‑ordered IDs. The high‑order bits (timestamp) make the IDs trend‑increasing, which reduces index fragmentation.

class SnowflakeSharding {
    final int nodeBits = 5;
    final long NODE_MASK = (1L << nodeBits) - 1;
    int shard(long snowId) { return (int) ((snowId >> 12) & NODE_MASK); }
}

Problems Introduced by Sharding

Distributed Transactions – two‑phase commit is costly; compensation mechanisms are preferred.

Cross‑Node JOIN – avoid MySQL’s multi‑database JOIN; use global tables, field redundancy, or application‑side assembly.

Cross‑Node Aggregation – must be performed in the application; pagination after large aggregations hurts performance.

Node Expansion & Data Migration – changing sharding rules after adding nodes requires data movement.

Migration Strategies

1. Full Migration with Downtime

Stop the service, run migration scripts, switch to the new sharding rule, then restart. This approach is risky for large‑scale data because most shards change.

2. Micro‑Migration without Downtime

Double the node count, add replica nodes (e.g., A2, B2) as slaves, sync data, then adjust the sharding rule from ID % 2 to ID % 4. After the new rule is live, break the old master‑slave links. Only half of the data needs to be migrated and the service stays online.

3. Power‑of‑Two Expansion

Capacity is planned in powers of two (4, 8, 32). When expanding to double the size, only half the data needs to move because the modulo operation can be implemented with a bitwise AND, which is highly efficient.

Practical References

MyCAT supports MySQL, SQL Server, Oracle, PostgreSQL, MongoDB and can be used for read‑write separation, sharding, and multi‑tenant architectures. Sharding‑JDBC is a lightweight Java framework that works with any JDBC‑compatible database; it supports flexible sharding strategies (equality, BETWEEN, IN, multi‑key) and transparent SQL parsing for aggregation, grouping, sorting, LIMIT, and Cartesian joins.

Sharding Key Design

1. Continuous (Range) Sharding

Shard by a specific field range (e.g., user ID, order time). Adding a new range after expansion does not require data migration, but hot‑spot skew can occur if recent data concentrates on a few nodes.

2. Modulo Sharding

Simple calculation shard = id % N. Expansion forces large‑scale data migration: when N doubles, roughly (N_new - N_old) / N_new of rows must be recomputed and moved. Range queries become inefficient because data is scattered across many shards.

3. Consistent‑Hash Sharding

Data movement on scaling is minimal; only the data on neighboring virtual nodes moves. Distribution is relatively balanced due to virtual nodes. Implementation is more complex than simple modulo.

4. Snowflake‑Based Sharding

Snowflake IDs are globally unique and monotonic, reducing index fragmentation. The algorithm depends on the system clock; clock rollback can cause duplicate IDs unless handled.

Node Expansion and Data Migration Details

Full Migration + Stop Service

Estimate migration time and announce downtime.

Stop the service and run migration scripts.

Switch to the new sharding rule.

Restart the service.

Micro‑Migration (Doubling Nodes)

Add two new databases (A2, B2) as slaves of existing masters (A → A2, B → B2) and synchronize data.

Adjust the sharding rule: ID % 2 = 0 becomes ID % 4 = 0 (A) and ID % 4 = 2 (A2); ID % 2 = 1 becomes ID % 4 = 1 (B) and ID % 4 = 3 (B2).

Break the master‑slave links, leaving four independent nodes.

Clean up redundant data at a convenient time.

Power‑of‑Two Planning

Large companies plan capacity in powers of two (e.g., 4, 8, 32). When expanding to double the size, only half of the data needs to be migrated because the modulo operation can be expressed as a bitwise AND, which is highly efficient.

Industrial‑Level Migration Workflow (Alibaba 100‑Billion‑Level Case)

Environment Preparation : configure online databases.

Full Sync : create full‑copy tasks for target tables.

Incremental Sync : after full copy, enable incremental replication with idempotent consumption.

Data Validation : verify consistency of full data.

Canary Test : replay traffic on pre‑release environment, validate cut‑over switches.

Second Validation : re‑validate and reconcile data.

Enable Dual Writes : write to both old and new databases.

Grey‑Scale Read : gradually route a fraction of reads ( userId % x) to the new cluster during low‑traffic periods.

Write‑Only New Cluster : switch all writes to the new cluster.

Migration Completion : monitor stability, then retire old cluster and release resources.