Rebalancing partitions

One of the biggest challenges with partitions is maintaining balance across shards.

Even if data is initially split evenly, imbalance will emerge over time.

Dataset size grows, causing some key ranges to hold more data than others.

Popular keys generate more traffic, creating hotspots.

Shard machines may fail, shifting load to remaining machines.

To reduce hotspots and ensure even distribution, shards must be rebalanced periodically.

A common strategy is to split a shard when it exceeds a configured size.

For example, a shard is split into two, each holding roughly half the data.

One shard remains on the original machine.

The other is moved to a new machine.

Splitting can be initiated manually by an administrator or automatically via data operations.

When a split occurs, the machine hosting the shard connects to the configuration service.

It updates metadata to reflect the new key range division.

Each shard now owns a range roughly half the size of the original.

After splitting, a rebalancing process can begin.

One or more shards are moved to new physical locations.

To move a shard, a clone is created on a new machine.

During cloning, the source shard handles all read and write requests.

Synchronization continues until the clone is fully up to date.

Once synchronized, the source machine notifies the configuration service.

The new shard location is registered.

All requests are then routed to the new machine.

The old copy is deleted on the source machine.

Splitting a shard involves only a metadata update.

It does not trigger immediate data migration.

Whether data migration occurs depends on a separate process.

A background balancer monitors shard distribution.

It tracks the number of shards per server.

If the difference between the largest and smallest server exceeds a threshold,

the balancer initiates shard migration.

The goal is to achieve equal shard counts across all servers.

Adding a new server triggers the balancer.

Shard migration begins to distribute load.

It may take time for shards to fully migrate.

Removing a server is the reverse of adding one.

Shards on the server to be removed are redistributed.

They are evenly spread across the remaining servers.

When data is deleted and a shard falls below a minimum size,

adjacent shards can be merged into one.

This reduces fragmentation and optimizes storage.

Splitting based on shard size is a strong design choice.

It keeps each shard below a configurable maximum size.

As data volume grows, more shards are created.

They are moved to optimize disk usage.

As request volume grows, new servers are added.

A fraction of the load is transferred to new machines.

This approach is used by databases like MongoDB and HBase.

Another option is to split one or more shards when a new server joins the cluster.

For example, two servers each hold two shards.

When a new server joins, existing shards are split to distribute load.

This ensures even distribution from the start.

It prevents overloading new servers.

It supports scalability without immediate imbalance.

The system randomly selects two shards to split.

The new server takes ownership of one half of each of these shards.

This approach is used by the Cassandra database.

Details of this method will be covered in the next video.

At the heart of both discussed rebalancing strategies is a splitting process.

This process prevents shards from growing too large.

It enables nearly unlimited scalability as data volume increases.

The splitting operation is not trivial.

When a shard is split, the system state changes.

All client applications must become aware of the new shard configuration.

This ensures clients write and read data from the correct set of new shards.

MongoDB stores shard metadata in a centralized metadata store.

If this store becomes unavailable, shard splits cannot occur.

Removing shard splitting may seem counterintuitive.

It implies shards can grow large.

But if the key range or hash range each shard owns is small from the start,

the shard size will never exceed manageable levels.

Even in worst-case scenarios, growth remains bounded.

This concept is worth investigating.

Each shard must own only a small subset of all possible keys.

This immediately requires a large initial number of shards.

For example, 1024 shards.

This number remains fixed throughout the system’s lifecycle.

All shards are assigned to a predefined set of machines.

For instance, 1664 shards per machine.

The key-to-shard mapping never changes.

Despite fixed assignments, load imbalance can still occur.

Some shards will own more keys than others.

Popular shards will generate disproportionately high load.

Shard movement can balance the load.

When a new machine joins the cluster,

each machine moves some of its shards to the new machine.

This redistributes shards until load is balanced.

This rebalancing method is called the fixed number of partition strategy.

Unlike previous strategies that change shard count,

this strategy keeps the number of shards constant.

This eliminates the need for shard splitting.

It simplifies implementation and maintenance.

Choosing the initial number of shards is difficult.

Too few shards lead to large shard sizes.

Too many shards complicate shard management.

Shards must be copied when new machines are added.

Shards must also be moved from machines undergoing maintenance.

Even with small initial ranges,

some shards may grow large.

This risk remains if key distribution is uneven.

Couchbase uses this rebalancing strategy.

Elasticsearch, a popular search engine, also uses it.

The size-based shard splitting strategy works well with both range and hash partitioning.

The join-based rebalancing strategy works best with hash partitioning.

The fixed number of shards strategy is typically used with lookup and hash partitioning.