The aggregation framework is MongoDB’s data processing pipeline — a chain of stages that transform documents sequentially. It powers analytics, reporting, data transformation, and complex joins.

Pipeline Concept 

  db.orders.aggregate([
  { $match: { status: "completed" } },     // Stage 1: filter
  { $group: { _id: "$userId", total: { $sum: "$amount" } } },  // Stage 2: aggregate
  { $sort: { total: -1 } },                // Stage 3: sort
  { $limit: 10 }                           // Stage 4: top 10
])

Each stage receives documents from the previous stage and passes output to the next. Order matters for performance.

Core Stages

$match — Filter Early 

  { $match: { status: "completed", createdAt: { $gte: ISODate("2024-01-01") } } }

Place $match as early as possible — it reduces documents flowing through the pipeline and can use indexes.

$project — Shape Documents 

  {
  $project: {
    name: 1,
    totalWithTax: { $multiply: ["$subtotal", 1.08] },
    year: { $year: "$createdAt" },
    _id: 0
  }
}

$group — Aggregate 

  {
  $group: {
    _id: { category: "$category", month: { $month: "$createdAt" } },
    totalRevenue: { $sum: "$amount" },
    orderCount: { $sum: 1 },
    avgOrder: { $avg: "$amount" },
    maxOrder: { $max: "$amount" }
  }
}

Group accumulators: $sum, $avg, $min, $max, $first, $last, $push, $addToSet.

$sort, $skip, $limit 

  { $sort: { totalRevenue: -1 } },
{ $skip: 0 },
{ $limit: 20 }

Prefer $limit before expensive stages when you only need top-N results.

$unwind — Flatten Arrays 

  // Input: { orderId: 1, items: [{ sku: "A" }, { sku: "B" }] }
{ $unwind: "$items" }
// Output: two documents, one per array element
  
  // Preserve empty arrays
{ $unwind: { path: "$items", preserveNullAndEmptyArrays: true } }

Joins with $lookup 

  db.orders.aggregate([
  { $match: { status: "completed" } },
  {
    $lookup: {
      from: "users",
      localField: "userId",
      foreignField: "_id",
      as: "user"
    }
  },
  { $unwind: "$user" },
  {
    $project: {
      orderId: 1,
      total: 1,
      userName: "$user.name",
      userEmail: "$user.email"
    }
  }
])

Correlated $lookup (MongoDB 3.6+) 

  {
  $lookup: {
    from: "inventory",
    let: { orderItems: "$items" },
    pipeline: [
      { $match: { $expr: { $in: ["$sku", "$$orderItems.sku"] } } },
      { $project: { sku: 1, stock: 1 } }
    ],
    as: "stockInfo"
  }
}

Advanced Stages

$facet — Multiple Pipelines in Parallel 

  db.products.aggregate([
  { $match: { category: "electronics" } },
  {
    $facet: {
      byBrand: [
        { $group: { _id: "$brand", count: { $sum: 1 } } },
        { $sort: { count: -1 } }
      ],
      priceStats: [
        { $group: {
            _id: null,
            avgPrice: { $avg: "$price" },
            minPrice: { $min: "$price" },
            maxPrice: { $max: "$price" }
        }}
      ],
      topRated: [
        { $sort: { rating: -1 } },
        { $limit: 5 },
        { $project: { name: 1, rating: 1 } }
      ]
    }
  }
])

One query, multiple analytics — ideal for dashboard APIs.

$bucket — Histogram Grouping 

  {
  $bucket: {
    groupBy: "$price",
    boundaries: [0, 25, 50, 100, 500],
    default: "500+",
    output: { count: { $sum: 1 }, products: { $push: "$name" } }
  }
}

$merge and $out — Write Results 

  // Write pipeline output to collection
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$userId", totalSpent: { $sum: "$amount" } } },
  { $merge: {
      into: "customer_stats",
      on: "_id",
      whenMatched: "replace",
      whenNotMatched: "insert"
  }}
])

Use $merge for incremental ETL; $out replaces the entire target collection.

Window Functions ($setWindowFields) 

  db.sales.aggregate([
  { $sort: { region: 1, date: 1 } },
  {
    $setWindowFields: {
      partitionBy: "$region",
      sortBy: { date: 1 },
      output: {
        runningTotal: {
          $sum: "$amount",
          window: { documents: ["unbounded", "current"] }
        },
        movingAvg: {
          $avg: "$amount",
          window: { documents: [-2, 0] }
        }
      }
    }
  }
])

Running totals, rankings, and moving averages without $lookup self-joins.

Real-World Example: Sales Dashboard 

  db.orders.aggregate([
  { $match: {
      status: "completed",
      createdAt: { $gte: ISODate("2024-01-01"), $lt: ISODate("2025-01-01") }
  }},
  { $unwind: "$items" },
  {
    $group: {
      _id: {
        month: { $month: "$createdAt" },
        category: "$items.category"
      },
      revenue: { $sum: { $multiply: ["$items.price", "$items.qty"] } },
      unitsSold: { $sum: "$items.qty" }
    }
  },
  { $sort: { "_id.month": 1, revenue: -1 } },
  {
    $group: {
      _id: "$_id.month",
      categories: {
        $push: { category: "$_id.category", revenue: "$revenue", units: "$unitsSold" }
      },
      monthTotal: { $sum: "$revenue" }
    }
  }
])

Pipeline Optimization

  1. $match first — filter before expensive stages; uses indexes
  2. $project early — reduce document size before $group or $lookup
  3. $limit before $lookup — if you only need top-N joined results
  4. allowDiskUse — for large sorts/groups exceeding 100 MB memory limit
  db.events.aggregate(pipeline, { allowDiskUse: true })
  1. Use $match not $filter when possible — $match uses indexes

explain() for Aggregations 

  db.orders.explain("executionStats").aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$userId", total: { $sum: "$amount" } } }
])

Check whether $match uses IXSCAN and whether $group spills to disk.

Common Mistakes

  • $lookup on large collections without prior $match — Cartesian explosion
  • $unwind on large arrays without $match first — multiplies document count
  • Forgetting $group requires _id field (use _id: null for global aggregation)
  • Using $skip for pagination in aggregation — same performance issues as find()
  • Not setting allowDiskUse for large analytics jobs — pipeline fails with memory error

Troubleshooting

Pipeline exceeds memory limit

  Exceeded memory limit for $group, but didn't allow external sort

Fix: { allowDiskUse: true } or add $match/$project to reduce data volume.

Slow $lookup

  • Ensure foreignField is indexed
  • Add $match before $lookup to reduce input documents
  • Consider embedding if join is always needed

Best Practices

  • Store pre-aggregated summaries for dashboards queried frequently
  • Use $merge for incremental nightly rollups
  • Test pipelines on production-size data, not dev samples
  • Version complex pipelines in application code or stored as views

What Comes Next

Data modeling decisions directly impact aggregation performance — embed vs reference determines whether you need $lookup at all.