⚙ MongoDB — Lesson 5: Aggregation Pipeline

A pipeline is an array of stage objects. Documents flow through each stage sequentially. Each stage can filter, transform, group, sort, or reshape the data.

// Basic pipeline structure
db.collection.aggregate([
  { $match: { /* filter */ } },          // Stage 1: filter
  { $group: { /* grouping */ } },          // Stage 2: group
  { $sort: { /* sort */ } },              // Stage 3: sort
  { $project: { /* reshape */ } },        // Stage 4: project
  { $limit: 10 }                          // Stage 5: limit
])

$match is like find() — it filters documents. Always put it first to use indexes and reduce pipeline input.

use salesdb

// Basic $match (equivalent to find filter)
db.sales.aggregate([
  { $match: { status: "completed" } }
])

// $match with date range
db.sales.aggregate([
  {
    $match: {
      status: "completed",
      saleDate: {
        $gte: new Date("2024-01-01"),
        $lt:  new Date("2024-02-01")
      }
    }
  }
])

// $match after $group (filters on aggregated results)
db.sales.aggregate([
  { $group: { _id: "$product", total: { $sum: "$amount" } } },
  { $match: { total: { $gt: 10000 } } }  // filter on aggregated total
])

$group groups documents by a key and computes aggregated values. The _id field specifies the group key — set to null to aggregate all documents.

// Sample data insert
db.sales.insertMany([
  { product: "Laptop",   category: "electronics", amount: 1200, qty: 2, rep: "Alice" },
  { product: "Mouse",    category: "electronics", amount: 30,   qty: 5, rep: "Bob"   },
  { product: "Desk",     category: "furniture",   amount: 600,  qty: 1, rep: "Alice" },
  { product: "Chair",    category: "furniture",   amount: 300,  qty: 3, rep: "Carol" },
  { product: "Monitor",  category: "electronics", amount: 450,  qty: 2, rep: "Bob"   },
  { product: "Keyboard", category: "electronics", amount: 80,   qty: 10, rep: "Carol" }
])

// Group by category — count and total
db.sales.aggregate([
  {
    $group: {
      _id: "$category",           // group key
      totalRevenue: { $sum: "$amount" },  // sum of amounts
      totalQty: { $sum: "$qty" },         // sum of quantities
      avgAmount: { $avg: "$amount" },      // average amount
      maxSale: { $max: "$amount" },        // largest single sale
      minSale: { $min: "$amount" },        // smallest single sale
      count: { $sum: 1 },                  // document count
      products: { $push: "$product" },    // array of product names
      firstRep: { $first: "$rep" }        // first rep encountered
    }
  }
])

// Group all documents (null key) — grand totals
db.sales.aggregate([
  {
    $group: {
      _id: null,
      grandTotal: { $sum: "$amount" },
      averageSale: { $avg: "$amount" },
      salesCount: { $sum: 1 }
    }
  }
])

// Group by multiple fields (compound key)
db.sales.aggregate([
  {
    $group: {
      _id: { category: "$category", rep: "$rep" },
      total: { $sum: "$amount" }
    }
  }
])

// Include/exclude fields (like projection in find)
db.sales.aggregate([
  { $project: { product: 1, amount: 1, _id: 0 } }
])

// Computed fields — create new fields on the fly
db.sales.aggregate([
  {
    $project: {
      product: 1,
      amount: 1,
      qty: 1,
      // New computed field: revenue = amount * qty
      revenue: { $multiply: ["$amount", "$qty"] },
      // Round to 2 decimal places
      pricePerUnit: { $round: [{ $divide: ["$amount", "$qty"] }, 2] },
      // String operations
      productUpper: { $toUpper: "$product" },
      label: { $concat: ["$product", " - $", { $toString: "$amount" }] },
      // Conditional
      isHighValue: { $gte: ["$amount", 500] }
    }
  }
])

// $project with date expressions
db.orders.aggregate([
  {
    $project: {
      orderId: "$_id",
      year: { $year: "$createdAt" },
      month: { $month: "$createdAt" },
      dayOfWeek: { $dayOfWeek: "$createdAt" }
    }
  }
])

// Sort by total revenue descending
db.sales.aggregate([
  { $group: { _id: "$rep", total: { $sum: "$amount" } } },
  { $sort: { total: -1 } }
])

// Top 3 best-selling reps (sorted + limited)
db.sales.aggregate([
  { $group: { _id: "$rep", total: { $sum: "$amount" } } },
  { $sort: { total: -1 } },
  { $limit: 3 }
])

// Pagination in aggregation
db.products.aggregate([
  { $match: { active: true } },
  { $sort: { name: 1 } },
  { $skip: 20 },
  { $limit: 10 }
])

$lookup is MongoDB's version of a SQL JOIN. It fetches documents from a foreign collection and embeds them as an array.

// Collections setup
// authors: { _id, name, email }
// posts: { _id, title, authorId, body }

// Basic $lookup — join posts with authors
db.posts.aggregate([
  {
    $lookup: {
      from: "authors",        // the foreign collection
      localField: "authorId", // field in the current collection
      foreignField: "_id",    // field in the foreign collection
      as: "authorData"        // name for the result array
    }
  }
])
// Result: each post gets an "authorData" array with matching author docs

// $lookup + $unwind to get single author object (not array)
db.posts.aggregate([
  {
    $lookup: {
      from: "authors",
      localField: "authorId",
      foreignField: "_id",
      as: "author"
    }
  },
  { $unwind: "$author" },   // flatten array to single object
  {
    $project: {
      title: 1,
      "author.name": 1,
      "author.email": 1
    }
  }
])

// Advanced $lookup with pipeline (more control)
db.orders.aggregate([
  { $match: { status: "completed" } },
  {
    $lookup: {
      from: "users",
      let: { userId: "$customerId" },    // bind local var
      pipeline: [
        { $match: { $expr: { $eq: ["$_id", "$$userId"] } } },
        { $project: { name: 1, email: 1, _id: 0 } }
      ],
      as: "customer"
    }
  },
  { $unwind: "$customer" }
])

// Without $unwind — tags is an array
// { _id:1, product:"Laptop", tags:["computer","portable"] }

// With $unwind — one doc per array element
db.products.aggregate([
  { $unwind: "$tags" }
])
// { _id:1, product:"Laptop", tags:"computer" }
// { _id:1, product:"Laptop", tags:"portable" }

// Count products per tag
db.products.aggregate([
  { $unwind: "$tags" },
  { $group: { _id: "$tags", count: { $sum: 1 } } },
  { $sort: { count: -1 } }
])

// $unwind options — preserve null/missing arrays
db.posts.aggregate([
  {
    $unwind: {
      path: "$comments",
      preserveNullAndEmptyArrays: true  // keep docs with no comments
    }
  }
])

// $addFields — add new fields without removing existing ones
db.products.aggregate([
  {
    $addFields: {
      totalValue: { $multiply: ["$price", "$stock"] },
      priceCategory: {
        $switch: {
          branches: [
            { case: { $lt: ["$price", 50] },   then: "budget" },
            { case: { $lt: ["$price", 200] },  then: "mid-range" },
            { case: { $lt: ["$price", 1000] }, then: "premium" }
          ],
          default: "luxury"
        }
      }
    }
  }
])

// $set is an alias for $addFields (newer, preferred)
db.products.aggregate([
  { $set: { discountedPrice: { $multiply: ["$price", 0.9] } } }
])

// $replaceRoot — make a subdocument the new root
db.users.aggregate([
  { $replaceRoot: { newRoot: "$profile" } }
])
// { firstName: "Alice", lastName: "Johnson", avatar: "..." }

// $replaceRoot with mergeObjects (keep _id)
db.users.aggregate([
  {
    $replaceRoot: {
      newRoot: { $mergeObjects: [{ _id: "$_id" }, "$profile"] }
    }
  }
])

$facet runs multiple aggregation pipelines simultaneously on the same input documents, returning a single document with results for each pipeline. Perfect for search results with faceted navigation.

// Get categories, price ranges, and top products in one query
db.products.aggregate([
  { $match: { active: true } },
  {
    $facet: {
      // Facet 1: count by category
      categoryCounts: [
        { $group: { _id: "$category", count: { $sum: 1 } } },
        { $sort: { count: -1 } }
      ],
      // Facet 2: price range distribution
      priceRanges: [
        {
          $bucket: {
            groupBy: "$price",
            boundaries: [0, 50, 200, 500, 2000],
            default: "Other",
            output: { count: { $sum: 1 } }
          }
        }
      ],
      // Facet 3: top rated products
      topRated: [
        { $sort: { rating: -1 } },
        { $limit: 5 },
        { $project: { name: 1, rating: 1, price: 1 } }
      ]
    }
  }
])

Monthly Sales Report

db.sales.aggregate([
  // Only completed sales from 2024
  {
    $match: {
      status: "completed",
      saleDate: { $gte: new Date("2024-01-01") }
    }
  },
  // Group by year and month
  {
    $group: {
      _id: {
        year: { $year: "$saleDate" },
        month: { $month: "$saleDate" }
      },
      revenue: { $sum: "$amount" },
      orders: { $sum: 1 },
      avgOrder: { $avg: "$amount" }
    }
  },
  // Add a formatted month label
  {
    $addFields: {
      monthLabel: {
        $dateToString: {
          format: "%Y-%m",
          date: {
            $dateFromParts: {
              year: "$_id.year",
              month: "$_id.month"
            }
          }
        }
      }
    }
  },
  { $sort: { "_id.year": 1, "_id.month": 1 } }
])

User Engagement Stats

db.users.aggregate([
  // Join with posts
  {
    $lookup: {
      from: "posts",
      localField: "_id",
      foreignField: "authorId",
      as: "posts"
    }
  },
  // Add computed fields
  {
    $addFields: {
      postCount: { $size: "$posts" },
      totalViews: { $sum: "$posts.views" },
      avgViews: { $avg: "$posts.views" }
    }
  },
  // Remove the full posts array from output
  { $project: { posts: 0, password: 0 } },
  // Only active authors with at least 1 post
  { $match: { postCount: { $gte: 1 } } },
  { $sort: { totalViews: -1 } }
])