MongoDB Aggregation Pipeline: Advanced Data Processing

MongoDB Aggregation Pipeline

Process and transform data with powerful aggregation operations.

Basic Pipeline Stages

// Match, Group, Sort
db.orders.aggregate([
    // Filter documents
    { $match: { status: 'completed', createdAt: { $gte: ISODate('2024-01-01') } } },

    // Group by customer
    { $group: {
        _id: '$customerId',
        totalOrders: { $sum: 1 },
        totalSpent: { $sum: '$amount' },
        avgOrder: { $avg: '$amount' }
    }},

    // Sort by total spent
    { $sort: { totalSpent: -1 } },

    // Limit results
    { $limit: 10 }
])

Joining Collections with $lookup

db.orders.aggregate([
    { $lookup: {
        from: 'customers',
        localField: 'customerId',
        foreignField: '_id',
        as: 'customer'
    }},
    { $unwind: '$customer' },
    { $project: {
        orderNumber: 1,
        amount: 1,
        'customer.name': 1,
        'customer.email': 1
    }}
])

Window Functions

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

Faceted Search

db.products.aggregate([
    { $facet: {
        'byCategory': [
            { $group: { _id: '$category', count: { $sum: 1 } } }
        ],
        'byPriceRange': [
            { $bucket: {
                groupBy: '$price',
                boundaries: [0, 50, 100, 500],
                output: { count: { $sum: 1 } }
            }}
        ],
        'topProducts': [
            { $sort: { sales: -1 } },
            { $limit: 5 }
        ]
    }}
])

Aggregation pipelines can handle complex analytics without moving data out of MongoDB.