MongoDB Lab Series: Ultra Advanced Analysis

# From the project root (stops the mongodb container from Tutorial 1)
docker compose down

# Verify it is gone
docker ps --filter name=mongodb

# Go to the project root first
cd big-data-analysis-with-mogodb   # or your own project folder
pwd

# Create the orchestration file
nano docker-compose.sharded.yml

# 1. Confirm nano wrote the file where Docker will look for it
pwd
ls docker-compose.sharded.yml

# 2. Dry-run it — Docker re-prints the config it parsed
docker compose -f docker-compose.sharded.yml config

# You should see the four services (cfg, shard1, shard2, mongos)
# and the three named volumes. An error here = invalid YAML, fix it first.

# 1. From the project root (where you saved the file), start everything
docker compose -f docker-compose.sharded.yml up -d

# 2. Wait ~10 seconds, then confirm all four containers are Up
docker compose -f docker-compose.sharded.yml ps
docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Ports}}" | grep -E "cfg|shard|mongos"

# 3. If one container is Restarting, its YAML line is wrong —
#    read its log and fix, then run the `up -d` command again:
#    docker compose -f docker-compose.sharded.yml logs shard1

# NOTE: `compose exec SERVICE ...` addresses the container by SERVICE name,
# so it works whether your containers are called `cfg` or `sharding-lab-cfg-1`.
# Run from the project root (where docker-compose.sharded.yml lives).

# 1. Initiate the config server replica set
docker compose -f docker-compose.sharded.yml exec cfg mongosh --port 27019 --quiet --eval \
  'rs.initiate({_id: "cfgRS", configsvr: true, members: [{_id: 0, host: "cfg:27019"}]})'

# 2. Initiate shard 1
docker compose -f docker-compose.sharded.yml exec shard1 mongosh --port 27018 --quiet --eval \
  'rs.initiate({_id: "shard1RS", members: [{_id: 0, host: "shard1:27018"}]})'

# 3. Initiate shard 2
docker compose -f docker-compose.sharded.yml exec shard2 mongosh --port 27018 --quiet --eval \
  'rs.initiate({_id: "shard2RS", members: [{_id: 0, host: "shard2:27018"}]})'

# Add shard 1
docker compose -f docker-compose.sharded.yml exec mongos mongosh --port 27017 --quiet --eval \
  'sh.addShard("shard1RS/shard1:27018")'

# Add shard 2
docker compose -f docker-compose.sharded.yml exec mongos mongosh --port 27017 --quiet --eval \
  'sh.addShard("shard2RS/shard2:27018")'

docker compose -f docker-compose.sharded.yml exec mongos mongosh --port 27017 --eval "sh.status()"

# All interaction from now on goes through the ROUTER, not the shards
docker compose -f docker-compose.sharded.yml exec mongos mongosh --port 27017

// 1. Enable sharding for the database
sh.enableSharding("olist_sharded")

// 2. Shard the orders collection on a hashed customer_id.
//    MongoDB needs an index supporting the shard key —
//    creating it up front avoids an error on import.
use olist_sharded
db.orders.createIndex({ customer_id: "hashed" })

// 3. Declare the sharded collection
sh.shardCollection("olist_sharded.orders", { customer_id: "hashed" })

// 4. Confirm it is registered as sharded
db.orders.getShardDistribution()   // empty for now — no data yet
sh.status({ verbose: false })

# Copy the orders CSV into the mongos container
# (compose cp uses the SERVICE name, so it works with prefixed container names too)
docker compose -f docker-compose.sharded.yml cp data/olist_dataset/olist_orders_dataset.csv mongos:/tmp/orders.csv

# Import through the router — port 27025 on localhost
docker compose -f docker-compose.sharded.yml exec mongos mongoimport \
  --db olist_sharded \
  --collection orders \
  --type csv --headerline \
  --file /tmp/orders.csv \
  --host localhost --port 27017

docker compose -f docker-compose.sharded.yml exec mongos mongosh --port 27017

use olist_sharded

// How many documents went to each shard?
db.orders.getShardDistribution()

// Total should match the standalone import (~99,441 orders)
db.orders.countDocuments()

// Confirm the collection reports itself as sharded
db.orders.stats().sharded   // true

[direct: mongos] olist_sharded> db.orders.getShardDistribution()
Shard shard1RS at shard1RS/shard1:27018
{
  data: '19.16MiB',
  docs: 49858,
  chunks: 2,
  'estimated data per chunk': '9.58MiB',
  'estimated docs per chunk': 24929
}
---
Shard shard2RS at shard2RS/shard2:27018
{
  data: '19.05MiB',
  docs: 49583,
  chunks: 2,
  'estimated data per chunk': '9.52MiB',
  'estimated docs per chunk': 24791
}
---
Totals
{
  data: '38.22MiB',
  docs: 99441,
  chunks: 4,
  'Shard shard1RS': [
    '50.13 % data',
    '50.13 % docs in cluster',
    '403B avg obj size on shard'
  ],
  'Shard shard2RS': [
    '49.86 % data',
    '49.86 % docs in cluster',
    '403B avg obj size on shard'
  ]
}

# 0. Find the Docker network YOUR cluster actually uses.
#    It is named after YOUR project folder (e.g. sharding-lab_default) —
#    do NOT hardcode big-data-analysis-with-mogodb_default unless that is your folder.
NET=$(docker inspect --format '{{range $k, $v := .NetworkSettings.Networks}}{{$k}}{{end}}' \
  $(docker compose -f docker-compose.sharded.yml ps -q mongos))
echo "Cluster network: $NET"

# 1. Start one more shard container attached to THAT network
docker run -d --name shard3 --network "$NET" \
  mongo:7 mongod --shardsvr --replSet shard3RS --port 27018 --bind_ip_all

# 2. Initiate its replica set
docker exec shard3 mongosh --port 27018 --quiet --eval \
  'rs.initiate({_id: "shard3RS", members: [{_id: 0, host: "shard3:27018"}]})'

# 2b. Wait until shard3 elects itself PRIMARY (must print 1) before registering it
sleep 5
docker exec shard3 mongosh --port 27018 --quiet --eval 'rs.status().myState'
# not 1 yet? wait a few seconds and repeat this line

# 3. Register it with the router — the balancer starts moving
#    chunks to the new shard automatically
docker compose -f docker-compose.sharded.yml exec mongos mongosh --port 27017 --quiet --eval \
  'sh.addShard("shard3RS/shard3:27018")'

# 4. Check the balance. Expectation check: our collection is only ~38 MiB in
#    total, so the gap between the fullest shard (19 MiB) and the empty one
#    (0 MiB) is ~19 MiB. The balancer treats a collection as balanced until
#    that gap reaches 3 x chunkSize (3 x 128 MB = 384 MB by default), so
#    shard3 stays EMPTY on its own — that is normal, not an error.
docker compose -f docker-compose.sharded.yml exec mongos mongosh --port 27017 --eval \
  'sh.status(); db.getSiblingDB("olist_sharded").orders.getShardDistribution()'
    

# 1. Cluster MEMBERSHIP — the authoritative list of shards.
#    This is the one that shows all three, even the empty one.
docker compose -f docker-compose.sharded.yml exec mongos mongosh --port 27017 --eval \
  'db.adminCommand({ listShards: 1 }).shards'

# 2. Same info, human-readable, with health state + balancer round
docker compose -f docker-compose.sharded.yml exec mongos mongosh --port 27017 --eval \
  'sh.status()'

# 3. DATA distribution — shard3RS is absent here, and that is expected:
#    it is registered, it just owns 0 chunks.
docker compose -f docker-compose.sharded.yml exec mongos mongosh --port 27017 --eval \
  'db.getSiblingDB("olist_sharded").orders.getShardDistribution()'

use olist_sharded

// 1. BEFORE — what is the chunk size? (no chunkSize field = the
//    128 MB global default) and how many chunks exist?
db.getSiblingDB("config").collections.findOne({ _id: "olist_sharded.orders" })
db.getSiblingDB("config").chunks.countDocuments({ ns: "olist_sharded.orders" })   // 4

// 2. Shrink the target chunk size for THIS collection only: 128 MB -> 1 MB
db.adminCommand({
  configureCollectionBalancing: "olist_sharded.orders",
  chunkSize: 1
})

// 3. Confirm it took effect (chunkSize: 1)
db.adminCommand({ configureCollectionBalancing: "olist_sharded.orders" })

// 4. IMPORTANT: changing chunkSize does NOT re-split chunks that
//    already exist — a cluster only splits when it must migrate.
//    So split it yourself, at real document keys:
const IDS = db.orders.distinct("customer_id").slice(0, 6)
IDS.forEach(id => sh.splitFind("olist_sharded.orders", { customer_id: id }))
// (a "would be a duplicate" error just means that key was already a
//  boundary — harmless, carry on)

// 5. Chunks are now much smaller and far more numerous
db.getSiblingDB("config").chunks.countDocuments({ ns: "olist_sharded.orders" })
sh.status()

// Pick a real chunk boundary to move. List the chunks with their
// min key, then hand one of them to shard3RS.
db.getSiblingDB("config").chunks
  .find({ ns: "olist_sharded.orders", shard: "shard1RS" }, { _id: 0, min: 1 })
  .toArray()

// Move the FIRST chunk of shard1RS over to shard3RS.
// (Use the exact `min` value printed above in place of MinKey.)
sh.moveChunk("olist_sharded.orders", MinKey, "shard3RS")

// Now the distribution finally shows all three shards
db.orders.getShardDistribution()

// Let the balancer keep working from here on
sh.startBalancer()
sh.isBalancerRunning()

// chunkSize: 0 removes the per-collection override and goes
// back to the global default (128 MB).
db.adminCommand({
  configureCollectionBalancing: "olist_sharded.orders",
  chunkSize: 0
})

db.adminCommand({ configureCollectionBalancing: "olist_sharded.orders" })

docker compose -f docker-compose.sharded.yml exec mongos mongosh --port 27017

use olist_sharded

// 1. distinct() — every distinct customer_id, deduped by the router.
//    No shard key in the filter, so this asks ALL shards and merges.
db.orders.distinct("customer_id")          // ~96,000 long array

// 2. Just a handful: .slice() keeps the output readable
db.orders.distinct("customer_id").slice(0, 5)

// 3. Grouping instead — the same idea, and $group collapses the
//    duplicates per shard before the router merges them.
db.orders.aggregate([
  { $group: { _id: "$customer_id" } },
  { $limit: 5 }
])

// 4. Stash one in a variable so every later command reuses the
//    same value instead of hand-copying an id.
const CID = db.orders.distinct("customer_id")[0]
print("Using customer_id:", CID)          // e.g. 9ef432ebbd257384d...

// 5. Sanity check: how many orders does this customer have?
//    ($match on the shard key -> only ONE shard is contacted)
db.orders.countDocuments({ customer_id: CID })

use olist_sharded

// TARGETED: filter on customer_id (the shard key).
// Look for "shards" in the output — only ONE shard is contacted.
db.orders.find({ customer_id: CID })
  .explain("executionStats")

// SCATTER-GATHER: filter on a non-shard-key field.
// "shards" now lists EVERY shard in the cluster.
db.orders.find({ order_status: "delivered" })
  .explain("executionStats")

// Aggregations route the same way — shard key in $match
// means only relevant shards run the pipeline:
db.orders.aggregate([
  { $match: { customer_id: CID } },          // same shard key -> 1 shard
  { $group: { _id: "$order_status", n: { $sum: 1 } } }
]).explain()

use olist_sharded

// 1. One insert. You did NOT say which shard should store it —
//    the router hashes customer_id and picks the owning shard.
db.orders.insertOne({
  order_id: "lab-0001",
  customer_id: "c-demo-0001",
  order_status: "created",
  order_total: 129.9
})

// 2. Bulk insert of 4 more. Each customer_id hashes independently,
//    so these 4 documents are most likely split across BOTH shards.
db.orders.insertMany([
  { order_id: "lab-0002", customer_id: "c-demo-0002", order_status: "created" },
  { order_id: "lab-0003", customer_id: "c-demo-0003", order_status: "shipped" },
  { order_id: "lab-0004", customer_id: "c-demo-0004", order_status: "shipped" },
  { order_id: "lab-0005", customer_id: "c-demo-0005", order_status: "delivered" }
])

// 3. How many did we just add? 99,441 -> 99,446
db.orders.countDocuments({ order_id: /^lab-/ })   // 5

// A. Filter on the SHARD KEY -> targeted, 1 shard.
//    CID is the customer_id you discovered in the step above.
db.orders.find({ customer_id: CID })
  .explain().queryPlanner.winningPlan.shards
// -> { shard1RS: {...} }   (one entry only)

// A2. The same customer's full order history, projected and sorted
db.orders.find({ customer_id: CID },
               { _id: 0, order_id: 1, order_status: 1, order_total: 1 })
  .sort({ order_id: 1 })

// B. Filter on a NON-shard-key field -> scatter-gather, every shard
db.orders.find({ order_id: "lab-0003" })
  .explain().queryPlanner.winningPlan.shards
// -> { shard1RS: {...}, shard2RS: {...} }

// C. Empty filter / counts -> always every shard
db.orders.countDocuments({})                 // 99,446
db.orders.distinct("order_status")            // ask all shards, merge + dedupe

// D. Same trick inside an aggregation: a $match on the shard key
//    prunes shards BEFORE the pipeline runs, so $group/$sort only
//    execute on the one shard that matters.
db.orders.aggregate([
  { $match: { customer_id: CID } },               // <- prunes to 1 shard
  { $group:   { _id: "$order_status", n: { $sum: 1 } } }
])

// D2. One-customer summary: order count and total spend
db.orders.aggregate([
  { $match: { customer_id: CID } },
  { $group: {
      _id: null,
      orders: { $sum: 1 },
      spent:  { $sum: "$order_total" },
      statuses: { $addToSet: "$order_status" }
  } }
])

// E. Remove the $match shard key and the same pipeline fans out
//    to every shard, then mongos merges partial results.
db.orders.aggregate([
  { $match: { order_status: "shipped" } },       // <- no pruning
  { $group:   { _id: "$customer_id", n: { $sum: 1 } } }
])

// 1. Targeted update: the filter carries the shard key, so the
//    router forwards the write to exactly one shard.
db.orders.updateOne(
  { customer_id: "c-demo-0001" },
  { $set: { order_status: "paid" } }
)

// 2. Scatter-gather update: no shard key in the filter, so EVERY
//    shard is asked to look for that document.
db.orders.updateMany(
  { order_status: "shipped" },
  { $set: { shipment_flag: true } }
)

// 3. Updating the SHARD KEY ITSELF. Since MongoDB 4.2 this is
//    ALLOWED, provided the filter pins the full shard key with an
//    equality match (and, for a non-null value, that the write runs
//    in a transaction or as a retryable write — mongosh uses
//    retryable writes by default, so this just works).
db.orders.updateOne(
  { customer_id: "c-demo-0001" },              // <- equality on the FULL shard key
  { $set: { customer_id: "c-demo-9999" } }
)
// -> { acknowledged: true, matchedCount: 1, modifiedCount: 1 }

//    THE CATCH: the document did NOT move. It is still physically on
//    the shard that held "c-demo-0001", but the config server now maps
//    "c-demo-9999" to the chunk owned by a DIFFERENT shard. Until the
//    balancer migrates it, a targeted query on the new key can come
//    back empty. Run this and watch:
//      db.orders.find({ customer_id: "c-demo-9999" })
db.orders.countDocuments({ customer_id: "c-demo-9999" })   // 0, or the doc if both keys hash to the same shard

//    The reliable way to MOVE a document: delete it and re-insert it.
//    The router then routes the insert to the correct shard itself.
db.orders.deleteOne({ customer_id: "c-demo-9999" })
db.orders.insertOne({ order_id: "lab-0001", customer_id: "c-demo-0001",
                      order_status: "paid", order_total: 129.9 })

// 4. Targeted delete (shard key present) vs scatter-gather delete
db.orders.deleteOne({ customer_id: "c-demo-0005" })   // 1 shard
db.orders.deleteMany({ order_status: "created" })     // every shard

// Remove everything this sub-section added, so later
// exercises and the verification checklist see the original
// 99,441 documents again.
db.orders.deleteMany({ order_id: /^lab-/ })
db.orders.countDocuments()          // 99,441 — back to the imported total