MongoDB Aggregation Pipeline

Build complex data processing pipelines for [Application] using MongoDB aggregation framework.

Requirements

1. Pipeline Stages

Implement aggregation for:

• [Use Case 1] - [Purpose] (e.g., Customer analytics)
• [Use Case 2] - [Purpose] (e.g., Sales reporting)
• [Use Case 3] - [Purpose] (e.g., Product metrics)

2. Stage Types to Use

Include stages for:

• $match - Filter documents
• $group - Aggregate data
• $project - Shape output
• $sort - Order results
• $lookup - Join collections
• $unwind - Deconstruct arrays
• $limit/$skip - Pagination

3. Aggregation Operations

Perform:

• Counting and summing
• Averaging and statistical calculations
• Grouping by multiple fields
• Date-based aggregations
• Text search integration

4. Performance Optimization

Ensure:

• Indexes support pipeline stages
• $match early in pipeline
• $project to reduce document size
• Appropriate use of $limit
• Pipeline complexity management

5. Complex Queries

Handle:

• Multi-collection joins ($lookup)
• Nested array operations
• Conditional logic ($cond, $switch)
• Date manipulations
• String operations

Implementation Pattern

import { MongoClient } from 'mongodb'

const client = new MongoClient(process.env.MONGODB_URI)
const db = client.db('[database]')

// Basic aggregation pipeline
const [pipelineName] = await db.collection('[collection]').aggregate([
  // Stage 1: Filter documents
  {
    $match: {
      [field]: [condition],
      [dateField]: { $gte: new Date('[start_date]') }
    }
  },
  
  // Stage 2: Group and aggregate
  {
    $group: {
      _id: '$[groupField]',
      total: { $sum: '$[sumField]' },
      average: { $avg: '$[avgField]' },
      count: { $sum: 1 },
      max: { $max: '$[maxField]' },
      min: { $min: '$[minField]' }
    }
  },
  
  // Stage 3: Sort results
  {
    $sort: { [sortField]: -1 }  // -1 for descending, 1 for ascending
  },
  
  // Stage 4: Limit results
  {
    $limit: [limit_number]
  },
  
  // Stage 5: Project (shape output)
  {
    $project: {
      [field1]: 1,
      [field2]: 1,
      [computedField]: { $multiply: ['$[field]', [value]] }
    }
  }
]).toArray()

// Lookup (join) example
const [joinedResults] = await db.collection('[collection1]').aggregate([
  {
    $lookup: {
      from: '[collection2]',
      localField: '[localField]',
      foreignField: '[foreignField]',
      as: '[resultField]'
    }
  },
  {
    $unwind: '$[resultField]'  // Deconstruct array if needed
  }
]).toArray()

// Date aggregation example
const [dateAggregation] = await db.collection('[collection]').aggregate([
  {
    $group: {
      _id: {
        year: { $year: '$[dateField]' },
        month: { $month: '$[dateField]' },
        day: { $dayOfMonth: '$[dateField]' }
      },
  {
    $sort: { '_id.year': 1, '_id.month': 1, '_id.day': 1 }
  }
])

Conditional Aggregation

db.collection('products').aggregate([
  {
    $addFields: {
      priceCategory: {
        $switch: {
          branches: [
            { case: { $lt: ['$price', 10] }, then: 'budget' },
            { case: { $lt: ['$price', 50] }, then: 'mid-range' }
          ],
          default: 'premium'
        }
      }
    }
  },
  {
    $group: {
      _id: '$priceCategory',
      count: { $sum: 1 },
      avgPrice: { $avg: '$price' }
    }
  }
])

Text Search Aggregation

db.collection('articles').aggregate([
  {
    $match: {
      $text: { $search: 'mongodb aggregation' }
    }
  },
  {
    $sort: { score: { $meta: 'textScore' } }
  },
  {
    $limit: 10
  }
])

Faceted Search

db.collection('products').aggregate([
  {
    $facet: {
      categoryCounts: [
        { $group: { _id: '$category', count: { $sum: 1 } } }
      ],
      priceRanges: [
        {
          $bucket: {
            groupBy: '$price',
            boundaries: [0, 25, 50, 100, 200],
            default: 'Other'
          }
        }
      ],
      topProducts: [
        { $sort: { sales: -1 } },
        { $limit: 5 }
      ]
    }
  }
])