MongoDB Sharding Overview

MongoDB Sharding

• Allows horizontal scaling for large workloads and data sets
• Employs a shared-nothing architecture, where nodes don't share resources
• Data is divided into shards, each functioning as a replica set for redundancy and availability
• Shards handle specific workloads, and partitions can be added or removed based on demand

Two Methods of Addressing System Growth

• Vertical Scaling: Expanding the capacity of a single server by upgrading CPU, RAM, and storage space
• Horizontal Scaling: Distributing the system's dataset and load across multiple servers, adding more servers as needed

Sharded Cluster Components

• Shard: Component of the sharded data, deployed as a replica set
• Mongos: Serves as a query router, connecting client applications to the sharded cluster
• Config Servers: Store cluster metadata and configuration parameters, must be a replica set (CSRS)

MongoDB Sharding Benefits

• Increased Read/Write Throughput: Achieved by distributing the data across multiple shards, allowing parallel processing. Example: Each shard can process 1,000 operations per second.
• Increased Storage Capacity: Adding a shard increases the total storage capacity. Example: If one shard is 4TB, adding another increases capacity by an additional 4TB.
• High Availability: Replica sets are essential for sharding, improving data availability by utilizing multiple servers.
• Data Locality: Zone sharding facilitates the design of distributed databases for geographically dispersed applications, supporting data residency in specific regions.

MongoDB Sharding Data Distribution

• Shard Key: Used to distribute a collection's documents across shards. Data is partitioned into non-overlapping intervals based on shard key values.
• Data Chunk Size: Maximum size of each data chunk is 128MB.
• Balancer: A background function that automatically migrates data chunks between shards to ensure equal data distribution.

Sharding Strategies

• Ranging Sharding: Divides data into ranges based on shard key values. Shard keys with similar values are likely to be stored in the same chunk, enabling focused operations.
• Hashed Sharding: Creates a hash of the shard key field's value. Data chunks are assigned ranges based on hashed shard key values, ensuring even data spread across shards.
• Zone Sharding: Separates data into distinct zones based on application requirements. Each zone can be associated with one or more shards and can store data from multiple zones.

How to Create a Sharded Collection in MongoDB

• Step 1: Set up Config Servers:

  • Start each configuration server replica set member with the --configsvr option.
  • Link to one of the replica set members using mongosh.
  • Use rs.initiate() on one of the replica set members to initialize the replica set.

• Step 2: Set up Shards:

  • Start each shard replica set member with the --shardsvr option and a unique replica set name.
  • Link to one of the replica set members using mongosh.
  • Use rs.initiate() on one of the replica set members, excluding the --configsvr option.

• Step 3: Start Mongos:

  • Configure Mongos and point it to the configuration server replica set using mongos --configdb.

• Step 4: Configure and Enable Sharding:

  • Connect to Mongos using mongosh.
  • Add shards to the cluster using sh.addShard().
  • Enable sharding for the database using sh.enableSharding().
  • Shard the collection with the sh.shardCollection() method. Choose between hashed sharding for even data distribution or range-based sharding for optimized distribution based on shard key values.

MongoDB Storage Systems

• Storage Engine: The primary component responsible for managing data.
• GridFS for Self-Managed Deployments: A storage system for handling files exceeding the 16MB document size limit.
• Journal: A write-ahead log system with checkpoints for data recovery after a shutdown.

WiredTiger Storage Engine

• Default Engine: Automatically selected unless otherwise specified. Supported in all MongoDB deployments.
• Operation & Limitations:
  • Uses optimistic concurrency control.
  • Transaction concurrency is dynamically optimized, with a limit of 128 read and 128 write transactions per node.
  • WiredTiger cache isn't partitioned, potentially degrading performance under heavy write workloads.
• Document Level Concurrency: Uses MultiVersion Concurrency Control (MVCC), allowing multiple clients to write to different documents simultaneously.
• Snapshots & Checkpoints: Snapshots provide consistent views of data at the start of operations. Checkpoints ensure data durability by writing consistent snapshots to disk every 60 seconds.
• Journal & Compression:
  • Journal persists all data modifications between checkpoints and uses compression by default.
  • Compression minimizes storage space, but requires additional CPU usage.
  • WiredTiger uses block compression for collections and prefix compression for indexes.
• Journal File Size Limit: Maximum journal file size is 100MB. Older files are automatically removed after recovery.