MongoDB Aggregation Pipeline

The aggregation pipeline processes documents through stages, transforming data at each step.

Basic Pipeline

db.orders.aggregate([
  // Stage 1: Filter documents
  { $match: { status: 'completed', createdAt: { $gte: new Date('2024-01-01') } } },

  // Stage 2: Group and calculate
  {
    $group: {
      _id: '$userId',
      totalSpent: { $sum: '$total' },
      orderCount: { $sum: 1 },
      avgOrderValue: { $avg: '$total' },
      lastOrder: { $max: '$createdAt' },
    }
  },

  // Stage 3: Filter grouped results
  { $match: { totalSpent: { $gte: 100 } } },

  // Stage 4: Sort
  { $sort: { totalSpent: -1 } },

  // Stage 5: Limit results
  { $limit: 10 },

  // Stage 6: Reshape output
  {
    $project: {
      userId: '$_id',
      _id: 0,
      totalSpent: { $round: ['$totalSpent', 2] },
      orderCount: 1,
      avgOrderValue: { $round: ['$avgOrderValue', 2] },
    }
  }
]);

$lookup (JOIN equivalent)

db.orders.aggregate([
  { $match: { status: 'pending' } },
  {
    $lookup: {
      from: 'users',
      localField: 'userId',
      foreignField: '_id',
      as: 'user',
    }
  },
  { $unwind: '$user' },  // Flatten user array to object
  {
    $project: {
      orderId: '$_id',
      userName: '$user.name',
      userEmail: '$user.email',
      total: 1,
      status: 1,
    }
  }
]);

// Advanced lookup with pipeline
db.orders.aggregate([
  {
    $lookup: {
      from: 'orderItems',
      let: { orderId: '$_id' },
      pipeline: [
        { $match: { $expr: { $eq: ['$orderId', '$orderId'] } } },
        { $lookup: { from: 'products', localField: 'productId', foreignField: '_id', as: 'product' } },
        { $unwind: '$product' },
        { $project: { name: '$product.name', quantity: 1, price: 1 } },
      ],
      as: 'items',
    }
  }
]);

$facet (Multiple Pipelines)

// Get results and counts in a single query
db.products.aggregate([
  { $match: { active: true } },
  {
    $facet: {
      // Branch 1: Paginated results
      results: [
        { $sort: { price: 1 } },
        { $skip: 0 },
        { $limit: 20 },
        { $project: { name: 1, price: 1, category: 1 } },
      ],
      // Branch 2: Count total
      totalCount: [
        { $count: 'count' }
      ],
      // Branch 3: Category breakdown
      byCategory: [
        { $group: { _id: '$category', count: { $sum: 1 } } },
        { $sort: { count: -1 } },
      ],
      // Branch 4: Price stats
      priceStats: [
        { $group: {
          _id: null,
          min: { $min: '$price' },
          max: { $max: '$price' },
          avg: { $avg: '$price' },
        }}
      ],
    }
  }
]);

Time-Series Aggregations

// Daily revenue for last 30 days
db.orders.aggregate([
  {
    $match: {
      createdAt: { $gte: new Date(Date.now() - 30 * 24 * 60 * 60 * 1000) },
      status: 'completed',
    }
  },
  {
    $group: {
      _id: {
        year: { $year: '$createdAt' },
        month: { $month: '$createdAt' },
        day: { $dayOfMonth: '$createdAt' },
      },
      revenue: { $sum: '$total' },
      orders: { $sum: 1 },
    }
  },
  { $sort: { '_id.year': 1, '_id.month': 1, '_id.day': 1 } },
  {
    $project: {
      date: {
        $dateToString: {
          format: '%Y-%m-%d',
          date: {
            $dateFromParts: {
              year: '$_id.year', month: '$_id.month', day: '$_id.day'
            }
          }
        }
      },
      revenue: { $round: ['$revenue', 2] },
      orders: 1,
    }
  }
]);

Text Search

// Create text index
db.articles.createIndex({ title: 'text', body: 'text' });

// Text search with scoring
db.articles.aggregate([
  { $match: { $text: { $search: 'mongodb performance optimization' } } },
  { $addFields: { score: { $meta: 'textScore' } } },
  { $sort: { score: { $meta: 'textScore' } } },
  { $limit: 10 },
]);

Use explain() with aggregations to understand query plans and ensure indexes are used.