Data sharding and partitioning strategies

Data Sharding and Partitioning Strategies

MongoDB achieves horizontal scaling through sharding technology, distributing large datasets across multiple servers. This mechanism addresses single-machine storage capacity and performance bottlenecks, with the core challenge being how to efficiently partition and route data.

Sharded Cluster Architecture

A typical sharded cluster consists of three roles:

1. mongos: The routing process responsible for forwarding client requests to the appropriate shard
2. config servers: Store cluster metadata and sharding configuration
3. shards: Actual mongod instances that store data

// Example of connecting to mongos
const { MongoClient } = require('mongodb');
const uri = "mongodb://mongos1:27017,mongos2:27017/?replicaSet=shardRs";
const client = new MongoClient(uri);

Shard Key Selection Strategies

The choice of shard key directly impacts query performance and cluster balance. Consider the following factors:

Cardinality Principle

High-cardinality fields are better suited as shard keys, such as user IDs or order numbers. Avoid low-cardinality fields like boolean values.

// Example of a poor shard key
{
  status: "active", // Only a few possible values
  created_at: ISODate()
}

Write Distribution Optimization

Avoid monotonically increasing shard keys (e.g., auto-incrementing IDs), which can cause "hot shard" issues. Consider these solutions:

• Composite keys: { userId: 1, timestamp: -1 }
• Hashed sharding: sh.shardCollection("db.users", { _id: "hashed" })

Sharding Strategy Types

Range Sharding

Divides data based on value ranges of the shard key, suitable for range query scenarios.

// Creating a range-sharded collection
sh.shardCollection("test.orders", { orderDate: 1 })

// Typical data distribution
shard1: orderDate 2020-01-01 ~ 2021-12-31
shard2: orderDate 2022-01-01 ~ 2023-12-31

Hashed Sharding

Evenly distributes data via hash functions, solving hot shard issues but sacrificing range query capabilities.

// Creating a hashed-sharded collection
sh.shardCollection("app.events", { deviceId: "hashed" })

// Example data distribution
shard1: Hash values 0x00000000 ~ 0x3FFFFFFF
shard2: Hash values 0x40000000 ~ 0x7FFFFFFF

Zone Sharding

Manually controls data distribution based on business rules, often used for multi-tenant or geographic partitioning scenarios.

// Creating zone rules
sh.addShardTag("shard1", "US-East")
sh.addShardTag("shard2", "EU-West")

// Adding zone ranges
sh.addTagRange("logs.traffic", 
  { region: "US", timestamp: MinKey },
  { region: "US", timestamp: MaxKey },
  "US-East"
)

Shard Management Operations

Sharded Cluster Monitoring

Use sh.status() to view shard status. Key metrics include:

• Data distribution balance
• Number of shard chunks
• Migration operation queue

// Checking shard status
db.adminCommand({ listShards: 1 })

// Viewing chunk distribution
use config
db.chunks.find({ ns: "products.items" })

Shard Rebalancing

Trigger the balancer when data distribution becomes uneven:

// Manually triggering balancing
sh.startBalancer()

// Setting balancing windows
use config
db.settings.update(
  { _id: "balancer" },
  { $set: { activeWindow: { start: "23:00", stop: "06:00" } } },
  { upsert: true }
)

Special Scenario Handling

Indexing Strategies for Sharded Collections

Each shard maintains its own indexes. Ensure:

• The shard key itself must be indexed
• Avoid globally unique indexes (unless they include the shard key)

// Correct index creation example
db.products.createIndex({ sku: 1, vendor: 1 })  // Compound index includes shard key

// Incorrect example (causes performance issues)
db.products.createIndex({ description: "text" }) // Full-text indexes are inefficient across shards

Sharded Transaction Limitations

Cross-shard transactions are only supported in MongoDB 4.2+ with the following constraints:

• Maximum 16MB data impact
• Recommended transaction duration under 1 second
• Requires WiredTiger storage engine for replica sets

// Cross-shard transaction example
const session = client.startSession();
try {
  session.startTransaction();
  await orders.insertOne({ _id: 1001, amount: 199 }, { session });
  await inventory.updateOne(
    { sku: "X100" }, 
    { $inc: { qty: -1 } },
    { session }
  );
  await session.commitTransaction();
} catch (error) {
  await session.abortTransaction();
}

Sharding Performance Optimization

Pre-Splitting Chunks

For known data distribution patterns, pre-split chunks to avoid automatic splitting overhead:

// Pre-splitting 100 chunks
sh.splitAt("db.collection", { userId: 0 });
for (let i = 1; i < 100; i++) {
  sh.splitAt("db.collection", { userId: i * 1000 });
}

Read/Write Concern Settings

Configure appropriate read/write concern levels in sharded environments:

// Strong consistency write operation
db.products.insert(
  { sku: "X200", price: 299 },
  { writeConcern: { w: "majority" } }
)

// Local read priority (reduces latency)
db.orders.find().readPref("nearest")

Sharding and Aggregation Pipelines

Aggregation operation execution strategies in sharded clusters:

• $match stages using the shard key first reduce shard scanning
• $lookup operations execute on mongos
• $group stages may encounter memory limit issues

// Optimized aggregation example
db.sales.aggregate([
  { $match: { 
    storeId: { $in: [101, 102] }, // Shard key field
    date: { $gt: ISODate("2023-01-01") }
  }},
  { $group: {
    _id: "$product",
    total: { $sum: "$amount" }
  }},
  { $sort: { total: -1 } }
])