Replica Sets lösen das Problem der Hochverfügbarkeit, aber nicht das Problem unbegrenzten Datenwachstums. Ein Replica Set kann vertikal skalieren – mehr RAM, schnellere CPUs, größere SSDs – aber es gibt physische Grenzen. Kein einzelner Server kann Petabytes an Daten halten oder Millionen Operations pro Sekunde bedienen. Selbst mit dem stärksten verfügbaren Hardware gibt es eine Obergrenze.
Sharding ist MongoDB’s Antwort auf horizontale Skalierung. Statt einen immer größeren Server zu kaufen, verteilt man Daten auf viele kleinere Server. Jeder Server – ein Shard – hält einen Teil der Daten. Zusammen können sie Datenmengen und Workloads bewältigen, die kein einzelner Server stemmen könnte. Dies ist fundamentale Skalierungsarchitektur für wirklich große Systeme: Social Networks mit Milliarden Usern, IoT-Plattformen mit Billionen Events, Analytics-Systeme mit Petabytes Historical Data.
Der Trade-off: Komplexität. Ein Sharded Cluster ist deutlich komplexer zu planen, aufzubauen und zu betreiben als ein Replica Set. Die Wahl des Shard Keys – des Feldes, nach dem Daten verteilt werden – ist kritisch und schwer zu ändern. Queries, die mehrere Shards betreffen, sind langsamer als lokale Queries. Sharding ist mächtig, aber nicht kostenlos. Die Entscheidung, wann zu sharden, ist eine der wichtigsten Architektur-Entscheidungen für MongoDB-Deployments.
Die Frage “Wann sollte ich sharden?” hat keine universelle Antwort, aber es gibt Indikatoren. Ein Replica Set zeigt Stress-Symptome, wenn es seine Kapazitätsgrenzen erreicht:
Speicherplatz-Erschöpfung: Die Daten wachsen schneller als man Storage hinzufügen kann. Oder die Datenbank ist bereits so groß, dass einzelne Server mit entsprechendem Storage unerschwinglich werden. Wenn die Datenbank 10 TB groß ist und 1 TB pro Monat wächst, ist die Schreibweise an der Wand.
RAM-Erschöpfung: Das Working Set – die häufig zugegriffenen Daten – passt nicht mehr in den RAM. Queries müssen zu Disk gehen, was Latenz um Größenordnungen erhöht. Man kann RAM upgraden, aber moderne Server toppen aus bei 1-2 TB. Wenn das Working Set 500 GB ist und steigt, ist RAM-Erschöpfung absehbar.
IOPS-Limits: Die Storage kann die I/O-Last nicht mehr bewältigen. SSDs haben IOPS-Limits, typischerweise zehntausende bis hunderttausende. Bei extremer Write-Heavy-Workload wird dies zum Bottleneck. Sharding verteilt I/O auf mehrere Storage-Systeme.
Single-Threaded Bottlenecks: Manche Operationen in MongoDB sind schwer parallelisierbar. Ein Sharded Cluster kann solche Operationen auf verschiedenen Shards parallel laufen lassen.
Die Faustregel: Sharding sollte in Betracht gezogen werden, wenn die Datenbank Hunderte Gigabytes oder Terabytes erreicht, oder wenn die Workload zehntausende Operations pro Sekunde übersteigt. Für kleinere Deployments ist ein gut konfiguriertes Replica Set oft ausreichend und deutlich simpler.
Ein MongoDB Sharded Cluster besteht aus drei Komponenten-Typen, die zusammenarbeiten, um Sharding transparent für Anwendungen zu machen.
Shards sind die Daten-Nodes. Jeder Shard ist ein Replica Set, das einen Teil der gesamten Daten hält. Ein Cluster mit drei Shards könnte die Daten in drei Teile partitionieren – etwa Users mit IDs 0-999999 auf Shard 1, 1000000-1999999 auf Shard 2, und 2000000+ auf Shard 3. Jeder Shard ist technisch ein vollwertiges Replica Set mit Primary und Secondaries. Dies garantiert Hochverfügbarkeit auf Shard-Ebene.
Config Servers speichern Metadaten über das Cluster: Welche Shards existieren, welche Collections sind gesharded, wie sind die Daten verteilt. Diese Metadaten sind kritisch – ohne sie weiß das Cluster nicht, wo Daten liegen. Config Servers laufen selbst als Replica Set (typischerweise drei Nodes) für Hochverfügbarkeit. Historisch konnten Config Servers ohne Replica Set laufen, aber seit MongoDB 3.4 ist das Replica Set Pflicht.
mongos (Router) sind stateless Proxy-Prozesse. Sie akzeptieren Client-Verbindungen, konsultieren die Config Servers, um zu wissen, wo Daten liegen, und routen Queries an die entsprechenden Shards. Ein mongos hält keine Daten selbst – er ist reine Routing-Logik. Anwendungen verbinden sich mit mongos, nicht direkt mit Shards. Dies macht Sharding transparent: Die App sieht eine “normale” MongoDB-Verbindung, der mongos managed die Komplexität.
Ein typisches Setup könnte sein: Drei Config Servers (als 3-Node-Replica Set), vier Shards (jeder ein 3-Node-Replica Set = 12 mongod-Prozesse), und zwei mongos-Router. Das sind bereits 17 Prozesse für ein minimales produktives Setup. Die operative Komplexität ist offensichtlich.
Der Shard Key ist das Feld (oder die Felder), nach dem MongoDB Daten partitioniert. Er bestimmt, auf welchem Shard ein Dokument landet. Die Wahl des Shard Keys ist fundamental – sie beeinflusst Performance, Skalierbarkeit und Query-Patterns für die Lebenszeit der Collection.
MongoDB teilt Daten in Chunks – kontinuierliche Bereiche von
Shard-Key-Werten. Ein Chunk könnte alle Dokumente mit
userId von 1000 bis 1999 enthalten. Chunks sind
typischerweise 64 MB oder 128 MB groß (konfigurierbar). MongoDB verteilt
Chunks auf Shards und versucht, die Verteilung balanciert zu halten.
Ein guter Shard Key hat drei Eigenschaften:
Hohe Cardinality: Viele mögliche Werte. Ein Shard
Key mit nur zwei Werten (etwa country: "US" oder
country: "non-US") limitiert auf zwei Chunks maximal – man
kann nicht auf mehr als zwei Shards verteilen. Ein userId
mit Millionen eindeutigen Werten erlaubt feine Granularität.
Gute Distribution: Werte sind gleichmäßig verteilt.
Ein Shard Key createdDate, wo 90% der Dokumente das heutige
Datum haben, führt zu Hotspots – ein Shard bekommt fast alle Writes. Ein
userId, wo User-IDs gleichmäßig über den Wertebereich
verteilt sind, balanciert gut.
Query-Isolation: Häufige Queries können auf einen
einzelnen Shard beschränkt werden. Wenn die App oft nach
userId filtert und userId der Shard Key ist,
kann mongos die Query direkt an den korrekten Shard routen. Queries ohne
Shard Key müssen an alle Shards gehen (Scatter-Gather), was langsamer
ist.
Diese drei Eigenschaften stehen oft in Konflikt. Ein perfekter Shard Key für Distribution könnte schlecht für Query-Isolation sein. Ein klassisches Beispiel:
Timestamp als Shard Key: Sehr schlechte Wahl. Neue Dokumente haben ähnliche Timestamps, landen also alle im selben Chunk, was zu einem “Hot Shard” führt – ein Shard bekommt alle Writes, die anderen sind idle. Distribution ist katastrophal.
User-ID als Shard Key: Gute Wahl für User-zentrische Anwendungen. User-IDs sind hochkardinal und meist gut verteilt. Queries wie “Finde alle Orders von User X” sind shard-isoliert. Aber Queries wie “Finde alle Orders der letzten Stunde” müssen alle Shards scannen.
Compound Shard Key (User-ID + Timestamp): Oft der beste Kompromiss. Die Leading-Komponente (User-ID) gibt Query-Isolation für User-spezifische Queries. Die Trailing-Komponente (Timestamp) fügt Granularität hinzu und vermeidet Hotspots bei Write-Bursts auf einzelne Users.
Die Wahl des Shard Keys ist schwer rückgängig zu machen. Vor MongoDB 5.0 war ein Shard Key permanent – ändern hieß die gesamte Collection exportieren, neu sharden, reimportieren. MongoDB 5.0 führte Resharding ein, aber es bleibt eine aufwendige Operation. Die initiale Wahl sollte deshalb sehr gründlich sein.
Wir bauen nun ein minimales Sharded Cluster lokal auf einem Host. Dies ist für Tests und Lernen gedacht – Produktion würde separate Hosts für jeden Shard verwenden. Unser Setup: Drei Config Servers, zwei Shards (jeder ein Single-Node-Replica Set für Einfachheit), und ein mongos.
Config Servers brauchen eigene Datenverzeichnisse und Ports. Wir starten drei:
# Verzeichnisse erstellen
mkdir -p ~/sharded-cluster/config1
mkdir -p ~/sharded-cluster/config2
mkdir -p ~/sharded-cluster/config3
# Config Server 1
mongod --configsvr --replSet configRS \
--port 27019 \
--dbpath ~/sharded-cluster/config1 \
--bind_ip localhost \
--fork \
--logpath ~/sharded-cluster/config1/mongod.log
# Config Server 2
mongod --configsvr --replSet configRS \
--port 27020 \
--dbpath ~/sharded-cluster/config2 \
--bind_ip localhost \
--fork \
--logpath ~/sharded-cluster/config2/mongod.log
# Config Server 3
mongod --configsvr --replSet configRS \
--port 27021 \
--dbpath ~/sharded-cluster/config3 \
--bind_ip localhost \
--fork \
--logpath ~/sharded-cluster/config3/mongod.logDie --configsvr-Flag teilt mongod mit, dass dies ein
Config Server ist, nicht ein normaler Daten-Server. Das
--replSet configRS definiert den Replica-Set-Namen.
Wir initialisieren das Config-Server-Replica Set:
mongosh --port 27019In der Shell:
rs.initiate({
_id: "configRS",
configsvr: true,
members: [
{ _id: 0, host: "localhost:27019" },
{ _id: 1, host: "localhost:27020" },
{ _id: 2, host: "localhost:27021" }
]
})Das configsvr: true-Flag ist kritisch – es markiert
dieses Replica Set als Config Server. Nach der Initiierung sollte
innerhalb von Sekunden ein Primary gewählt sein.
Jeder Shard ist ein Replica Set. Für unser lokales Setup verwenden wir Single-Node-Replica Sets – in Produktion wären das 3+-Node-Sets.
Shard 1:
mkdir -p ~/sharded-cluster/shard1
mongod --shardsvr --replSet shard1RS \
--port 27030 \
--dbpath ~/sharded-cluster/shard1 \
--bind_ip localhost \
--fork \
--logpath ~/sharded-cluster/shard1/mongod.logInitialisieren:
mongosh --port 27030rs.initiate({
_id: "shard1RS",
members: [{ _id: 0, host: "localhost:27030" }]
})Shard 2:
mkdir -p ~/sharded-cluster/shard2
mongod --shardsvr --replSet shard2RS \
--port 27040 \
--dbpath ~/sharded-cluster/shard2 \
--bind_ip localhost \
--fork \
--logpath ~/sharded-cluster/shard2/mongod.logInitialisieren:
mongosh --port 27040rs.initiate({
_id: "shard2RS",
members: [{ _id: 0, host: "localhost:27040" }]
})Die --shardsvr-Flag markiert diese mongod-Instanzen als
Shard-Server. Dies hat subtile Effekte – etwa unterschiedliche
Default-Chunk-Größen und spezielles Verhalten beim Balancing.
Der mongos-Prozess ist kein mongod – er persistiert keine Daten. Er braucht nur die Config-Server-Connection-String:
mongos --configdb configRS/localhost:27019,localhost:27020,localhost:27021 \
--port 27017 \
--bind_ip localhost \
--fork \
--logpath ~/sharded-cluster/mongos.logDer --configdb-Parameter spezifiziert das
Config-Server-Replica Set. Das Format ist
<replicaSetName>/<host1>,<host2>,<host3>.
mongos verbindet sich mit den Config Servers, liest die
Cluster-Metadaten und ist ready, Queries zu routen.
Wir verbinden uns mit mongos (nicht mit einem Shard!):
mongosh --port 27017Von außen sieht dies aus wie eine normale MongoDB-Verbindung. Aber intern routet mongos zu Shards.
Mit mongos verbunden, fügen wir die Shards hinzu:
sh.addShard("shard1RS/localhost:27030")
sh.addShard("shard2RS/localhost:27040")Diese Commands registrieren die Shard-Replica Sets beim Cluster. MongoDB speichert diese Information in den Config Servers. Wir verifizieren:
sh.status()Die Output zeigt:
{
shardingVersion: { ... },
shards: {
_id: 'shard1RS',
host: 'shard1RS/localhost:27030',
state: 1
},
{
_id: 'shard2RS',
host: 'shard2RS/localhost:27040',
state: 1
},
// ...
}Beide Shards sind registered und state 1 (active). Das Cluster ist bereit, Daten zu sharden.
Ein Sharded Cluster kann normale (ungesharded) und geshardete Datenbanken mischen. Um eine Datenbank zu sharden:
sh.enableSharding("testDB")Dies markiert die Datenbank als shard-fähig. Collections innerhalb dieser Datenbank können nun gesharded werden. Wichtig: Die Datenbank selbst ist nicht gesharded – nur spezifische Collections innerhalb der Datenbank sind es.
Wir erstellen eine Collection und sharden sie:
use testDB
// Collection mit Daten füllen (für Demo-Zwecke)
for (let i = 0; i < 10000; i++) {
db.users.insertOne({
userId: i,
name: `User${i}`,
email: `user${i}@example.com`,
createdAt: new Date()
})
}
// Collection sharden mit userId als Shard Key
sh.shardCollection("testDB.users", { userId: 1 })Der sh.shardCollection-Command definiert den Shard Key.
Hier ist es { userId: 1 } – ein aufsteigender Index auf
userId. MongoDB erstellt automatisch einen Index auf dem
Shard Key, falls er nicht existiert.
Nach dem Sharden beginnt MongoDB, Chunks zu erstellen und auf Shards zu verteilen. Initial sind alle Daten oft auf einem Shard (dem, der sie ursprünglich hielt). Der Balancer – ein Hintergrund-Prozess – wird Chunks graduell auf die Shards verteilen.
Wir können die Verteilung sehen:
db.users.getShardDistribution()Die Output zeigt, wie viele Dokumente und wie viel Speicher auf jedem Shard liegen. Initial könnte es unbalanciert sein, aber nach Minuten sollte es sich ausgleichen.
MongoDB verteilt Daten in Chunks – kontinuierliche Bereiche des
Shard-Key-Wertebereichs. Ein Chunk könnte userId von 0 bis
999 enthalten. Chunks haben eine Größe-Limit (Default 128 MB seit
Version 6.0). Wächst ein Chunk über das Limit, splittet MongoDB ihn in
zwei kleinere Chunks.
Der Balancer ist ein Hintergrund-Prozess, der auf dem Primary des Config-Server-Replica-Sets läuft. Er monitort die Chunk-Verteilung und bewegt Chunks zwischen Shards, um Balance zu halten. Ein Shard, der signifikant mehr Chunks hat als andere, wird Chunks an weniger belastete Shards abgeben.
Chunk-Migrationen sind live – die Daten bleiben verfügbar während der Migration. Der Prozess:
Dieser Prozess ist normalerweise transparent, kann aber bei sehr großen Chunks oder langsamen Netzwerken Minuten oder Stunden dauern. Während der Migration ist Performance leicht beeinträchtigt, aber das System bleibt funktional.
Der Balancer läuft standardmäßig kontinuierlich, kann aber für Wartungsfenster gestoppt werden:
sh.stopBalancer()
sh.getBalancerState() // falseFür Produktions-Deployments ist es oft sinnvoll, Balancing auf Zeitfenster mit geringer Last zu beschränken:
use config
db.settings.updateOne(
{ _id: "balancer" },
{ $set: {
activeWindow: {
start: "02:00",
stop: "06:00"
}
}},
{ upsert: true }
)Dies limitiert Balancing auf 02:00-06:00 Uhr. Außerhalb dieses Fensters passiert kein Chunk-Movement, was Haupt-Traffic-Zeiten nicht beeinträchtigt.
Wenn eine Query den Shard Key enthält, kann mongos sie direkt an den korrekten Shard routen. Dies nennt man “targeted query”:
db.users.find({ userId: 5000 })mongos weiß (durch Konsultation der Config Servers), dass
userId: 5000 auf Shard X liegt, und routet die Query nur
dorthin. Dies ist fast so schnell wie eine Query auf einem ungesharded
System.
Queries ohne Shard Key müssen an alle Shards gehen:
db.users.find({ email: "user5000@example.com" })mongos sendet diese Query an alle Shards (Scatter), jeder Shard führt sie lokal aus und returned Results, mongos merged die Results (Gather). Dies ist langsamer – die Latenz ist mindestens die des langsamsten Shards, und der Overhead des Merging addiert sich.
Die Explain-Output zeigt, welche Shards involviert waren:
db.users.find({ email: "user5000@example.com" }).explain()Das shards-Feld listet alle beteiligten Shards. Für
Performance-kritische Queries sollte man Shard-Key-Inklusivität
anstreben.
Vor MongoDB 5.0 war der Shard Key unveränderlich. Stellte sich heraus, dass die Wahl suboptimal war, blieb nur die manuelle Migration: Daten exportieren, Collection neu sharden, Daten reimportieren. MongoDB 5.0 führte Resharding ein – die Fähigkeit, den Shard Key einer bestehenden geshardeten Collection zu ändern.
db.adminCommand({
reshardCollection: "testDB.users",
key: { userId: 1, createdAt: 1 } // Neuer compound key
})MongoDB reorganisiert die Daten im Hintergrund. Während des Reshardings bleibt die Collection verfügbar, aber die Performance kann beeinträchtigt sein. Bei großen Collections kann Resharding Stunden oder Tage dauern.
Resharding ist eine schwere Operation und sollte nicht leichtfertig durchgeführt werden. Aber es macht Sharding weniger beängstigend – die Entscheidung ist nicht mehr absolut final.
Der Hauptvorteil von Sharding ist fast unbegrenzte Skalierbarkeit. Wenn zwei Shards an ihre Kapazitätsgrenze kommen, fügt man einen dritten hinzu:
# Shard 3 Replica Set starten
mkdir -p ~/sharded-cluster/shard3
mongod --shardsvr --replSet shard3RS \
--port 27050 \
--dbpath ~/sharded-cluster/shard3 \
--bind_ip localhost \
--fork \
--logpath ~/sharded-cluster/shard3/mongod.log
# Initialisieren
mongosh --port 27050
rs.initiate({
_id: "shard3RS",
members: [{ _id: 0, host: "localhost:27050" }]
})Dem Cluster hinzufügen:
// Mit mongos verbunden
sh.addShard("shard3RS/localhost:27050")MongoDB erkennt den neuen Shard und der Balancer beginnt automatisch, Chunks auf ihn zu verteilen. Innerhalb von Stunden ist die Last gleichmäßiger verteilt. Dies kann wiederholt werden – theoretisch unbegrenzt. Produktive Deployments mit Dutzenden oder Hunderten Shards existieren.
Die folgende Tabelle fasst die wichtigsten Sharding-Commands zusammen:
| Command | Zweck | Typisches Szenario |
|---|---|---|
sh.addShard() |
Shard hinzufügen | Cluster erweitern |
sh.enableSharding() |
Datenbank für Sharding aktivieren | Vor Collection-Sharding |
sh.shardCollection() |
Collection sharden | Neue geshardete Collection |
sh.status() |
Cluster-Status anzeigen | Health-Check, Debugging |
sh.stopBalancer() |
Balancer stoppen | Wartungsfenster |
sh.startBalancer() |
Balancer starten | Nach Wartung |
db.collection.getShardDistribution() |
Datenverteilung anzeigen | Performance-Analyse |
db.adminCommand({reshardCollection}) |
Shard Key ändern | Shard-Key-Korrektur |
Sharding ist MongoDB’s Mechanismus für unbegrenzte Skalierung, aber es kommt mit Komplexität. Die Wahl des Shard Keys ist kritisch und schwer zu ändern. Die Architektur – Config Servers, Shards, mongos – ist multi-dimensional und erfordert sorgfältige Planung. Queries ohne Shard-Key-Inclusivität sind langsamer. Aber für wirklich große Systeme gibt es keine Alternative. Wenn ein Replica Set an seine Grenzen stößt, ist Sharding der nächste Schritt. Die operative Investition zahlt sich aus durch praktisch unbegrenzte Skalierbarkeit.