Aggregation Pipeline

The aggregation pipeline is MongoDB's framework for data processing and transformation. It passes documents through a series of stages, each transforming its input and passing results to the next stage — similar to a Unix pipe.

Pipeline vs find()

Feature	find()	Aggregation
Filtering	Yes	$match
Projection	Yes	$project
Sorting	Yes	$sort
Grouping	No	$group
Joining collections	No	$lookup
Computed fields	Limited	$addFields / $project
Writing results	No	$out / $merge

Basic Pipeline Structure

Pipeline syntax

db.collection.aggregate([
  { $match:   { /* filter */  } },
  { $group:   { /* group  */  } },
  { $sort:    { /* sort   */  } },
  { $limit:   10               },
  { $project: { /* shape  */  } },
])

A Complete Example

This pipeline finds the top 5 product categories by average rating, requiring at least 10 reviews.

Top-rated categories

db.products.aggregate([
  // 1. Only active products
  { $match: { active: true } },

  // 2. Group by category
  {
    $group: {
      _id: '$category',
      avgRating:   { $avg: '$rating' },
      reviewCount: { $sum: 1 },
    },
  },

  // 3. Keep categories with ≥ 10 reviews
  { $match: { reviewCount: { $gte: 10 } } },

  // 4. Best-rated first
  { $sort: { avgRating: -1 } },

  // 5. Top 5 only
  { $limit: 5 },

  // 6. Clean up output field names
  {
    $project: {
      _id: 0,
      category:    '$_id',
      avgRating:   { $round: ['$avgRating', 2] },
      reviewCount: 1,
    },
  },
])

Place $match Early

Always place $match as early as possible in the pipeline. Early filtering reduces the documents subsequent stages must process. When $match is the first stage and its fields are indexed, MongoDB uses the index — exactly like find().

Tip

A $match in position 1 that hits an index can reduce 1 000 000 documents to 50 before $group ever runs.

Memory Limits and allowDiskUse

Each pipeline stage is limited to 100 MB of RAM. For larger datasets enable disk spill with allowDiskUse: true.

Allow disk use for large aggregations

db.orders.aggregate(
  [ { $group: { _id: '$customerId', total: { $sum: '$amount' } } } ],
  { allowDiskUse: true }
)

Explain an Aggregation

explain() on a pipeline

db.orders.explain('executionStats').aggregate([
  { $match: { status: 'completed' } },
  { $group: { _id: '$customerId', count: { $sum: 1 } } },
])

Monthly Revenue Report

Real-world: monthly revenue

db.orders.aggregate([
  {
    $match: {
      createdAt: { $gte: new Date('2024-01-01') },
      status: 'paid',
    },
  },
  {
    $group: {
      _id: { $month: '$createdAt' },
      revenue: { $sum: '$total' },
      orders:  { $sum: 1 },
    },
  },
  { $sort: { _id: 1 } },
  {
    $project: {
      _id: 0,
      month:   '$_id',
      revenue: 1,
      orders:  1,
    },
  },
])

Cursor result

aggregate() returns a cursor just like find(). Chain .toArray() or .forEach() to consume it.

Warning

Avoid $unwind on large arrays without a following $group or $limit — it multiplies document count and can exhaust the 100 MB stage memory limit.