MongoDB Aggregation Pipeline

Master powerful data transformation with aggregation pipeline.

Quick Start

Basic Pipeline Structure

const result = await collection.aggregate([
  { $match: { status: 'active' } },
  { $group: { _id: '$category', count: { $sum: 1 } } },
  { $sort: { count: -1 } },
  { $limit: 10 }
]).toArray();

Common Pipeline Stages

// $match: Filter documents (like WHERE in SQL)
{ $match: { age: { $gte: 18 }, status: 'active' } }

// $group: Group documents and aggregate
{ $group: {
  _id: '$city',
  total: { $sum: '$amount' },
  average: { $avg: '$price' },
  count: { $sum: 1 }
}}

// $project: Transform fields
{ $project: {
  name: 1,
  email: 1,
  fullName: { $concat: ['$firstName', ' ', '$lastName'] },
  _id: 0
}}

// $sort: Sort results
{ $sort: { createdAt: -1 } }  // -1 for descending, 1 for ascending

// $limit and $skip: Pagination
{ $limit: 10 }
{ $skip: 20 }

// $unwind: Deconstruct arrays
{ $unwind: '$tags' }  // One document per tag

// $lookup: Join collections
{ $lookup: {
  from: 'categories',
  localField: 'categoryId',
  foreignField: '_id',
  as: 'category'
}}

// $facet: Multi-faceted search
{ $facet: {
  byCategory: [{ $group: { _id: '$category', count: { $sum: 1 } } }],
  byPrice: [{ $group: { _id: null, avg: { $avg: '$price' } } }]
}}

Aggregation Functions

// Numeric functions
{ $sum: 1 }                    // Count
{ $sum: '$amount' }            // Sum field
{ $avg: '$price' }             // Average
{ $min: '$quantity' }          // Minimum
{ $max: '$quantity' }          // Maximum

// Array functions
{ $push: '$tags' }             // Collect all values
{ $addToSet: '$category' }     // Collect unique values
{ $first: '$name' }            // First element
{ $last: '$name' }             // Last element

// String functions
{ $concat: ['$firstName', ' ', '$lastName'] }
{ $substr: ['$email', 0, 5] }
{ $toLower: '$name' }
{ $toUpper: '$name' }

// Conditional
{ $cond: [
  { $gte: ['$age', 18] },
  'Adult',
  'Minor'
]}

Real-World Examples

Sales Report by Category

await orders.aggregate([
  { $match: { status: 'completed' } },
  { $group: {
    _id: '$category',
    totalSales: { $sum: '$amount' },
    ordersCount: { $sum: 1 },
    avgOrderValue: { $avg: '$amount' }
  }},
  { $sort: { totalSales: -1 } }
]).toArray();

User Activity Summary

await users.aggregate([
  { $match: { lastActive: { $gte: new Date(Date.now() - 30*24*60*60*1000) } } },
  { $project: {
    name: 1,
    email: 1,
    lastActive: 1,
    daysInactive: {
      $floor: {
        $divide: [
          { $subtract: [new Date(), '$lastActive'] },
          1000 * 60 * 60 * 24
        ]
      }
    }
  }},
  { $sort: { daysInactive: 1 } }
]).toArray();

Join with Lookup

await orders.aggregate([
  { $match: { status: 'pending' } },
  { $lookup: {
    from: 'customers',
    localField: 'customerId',
    foreignField: '_id',
    as: 'customer'
  }},
  { $unwind: '$customer' },
  { $project: {
    orderId: '$_id',
    customerName: '$customer.name',
    total: '$amount',
    _id: 0
  }}
]).toArray();

Python Example

from pymongo import MongoClient

client = MongoClient('mongodb://localhost:27017')
collection = client['db']['collection']

pipeline = [
    {'$match': {'status': 'active'}},
    {'$group': {'_id': '$category', 'count': {'$sum': 1}}},
    {'$sort': {'count': -1}}
]

results = list(collection.aggregate(pipeline))
for doc in results:
    print(doc)

Performance Tips

✅ Use $match early to filter documents ✅ Avoid $project before necessary stages ✅ Use indexes on $match fields ✅ Use $limit before expensive operations ✅ Monitor aggregation performance ✅ Use $facet for parallel processing ✅ Avoid large $unwind operations