find() gets documents. The aggregation pipeline transforms them. It is MongoDB’s answer to SQL GROUP BY, JOIN, and window functions.

How the Pipeline Works

Documents flow through a sequence of stages. Each stage takes the output of the previous one as input.

db.orders.aggregate([
  { $match:   { status: "delivered" } },       // Stage 1: filter
  { $group:   { _id: "$user_id", total: { $sum: "$amount" } } }, // Stage 2: group
  { $sort:    { total: -1 } },                 // Stage 3: sort
  { $limit:   10 }                             // Stage 4: limit
])

$match — Filter Documents

// Equivalent to WHERE in SQL
db.orders.aggregate([
  {
    $match: {
      status: "delivered",
      createdAt: { $gte: ISODate("2026-01-01"), $lt: ISODate("2027-01-01") }
    }
  }
])

Put $match as early as possible. It reduces the number of documents processed by later stages.

$group — Aggregate

// Total revenue per user
db.orders.aggregate([
  { $match: { status: "delivered" } },
  {
    $group: {
      _id: "$user_id",
      order_count: { $sum: 1 },
      total_spent:  { $sum: "$amount" },
      avg_order:    { $avg: "$amount" },
      first_order:  { $min: "$createdAt" },
      last_order:   { $max: "$createdAt" }
    }
  },
  { $sort: { total_spent: -1 } }
])

Common $group accumulators:

AccumulatorWhat it does
$sumSum of values
$avgAverage
$minMinimum value
$maxMaximum value
$countCount documents (MongoDB 5+)
$pushCollect values into array
$addToSetCollect unique values
$first / $lastFirst or last value in group

$lookup — Join Collections

// Join orders with users (left outer join)
db.orders.aggregate([
  {
    $lookup: {
      from: "users",        // collection to join
      localField: "user_id", // field in orders
      foreignField: "_id",  // field in users
      as: "user"            // output array field name
    }
  },
  { $unwind: "$user" },  // flatten the array (1:1 join)
  {
    $project: {
      orderId: "$_id",
      amount: 1,
      userName: "$user.name",
      userEmail: "$user.email"
    }
  }
])

$unwind — Flatten Arrays

// Each tag gets its own document
db.posts.aggregate([
  { $unwind: "$tags" },
  {
    $group: {
      _id: "$tags",
      post_count: { $sum: 1 }
    }
  },
  { $sort: { post_count: -1 } }
])
// Result: most popular tags with their post counts

$project — Reshape Documents

db.orders.aggregate([
  {
    $project: {
      _id: 0,                          // exclude _id
      order_id: "$_id",                // rename field
      amount: 1,                       // include
      year: { $year: "$createdAt" },   // computed field
      month: { $month: "$createdAt" },
      total_with_tax: { $multiply: ["$amount", 1.19] }
    }
  }
])

$addFields — Add Computed Fields

db.products.aggregate([
  {
    $addFields: {
      discounted_price: {
        $multiply: ["$price", { $subtract: [1, "$discount_rate"] }]
      },
      in_stock: { $gt: ["$stock", 0] }
    }
  }
])

Real Example: Monthly Revenue Report

db.orders.aggregate([
  // Only count paid orders
  { $match: { status: { $in: ["delivered", "processing"] } } },

  // Add year and month fields
  {
    $addFields: {
      year:  { $year: "$createdAt" },
      month: { $month: "$createdAt" }
    }
  },

  // Group by year + month
  {
    $group: {
      _id: { year: "$year", month: "$month" },
      revenue:     { $sum: "$amount" },
      order_count: { $sum: 1 },
      avg_order:   { $avg: "$amount" }
    }
  },

  // Sort chronologically
  { $sort: { "_id.year": 1, "_id.month": 1 } },

  // Rename output fields
  {
    $project: {
      _id: 0,
      year:        "$_id.year",
      month:       "$_id.month",
      revenue:     { $round: ["$revenue", 2] },
      order_count: 1,
      avg_order:   { $round: ["$avg_order", 2] }
    }
  }
])

Node.js Example

import mongoose, { Schema, model } from "mongoose";

// Order model (define schema before using it)
const orderSchema = new Schema({
  status: String,
  amount: Number,
  createdAt: Date,
});
const Order = model("Order", orderSchema);

interface RevenueReport {
  year: number;
  month: number;
  revenue: number;
  order_count: number;
}

async function getMonthlyRevenue(year: number): Promise<RevenueReport[]> {
  return Order.aggregate([
    {
      $match: {
        status: { $in: ["delivered", "processing"] },
        createdAt: {
          $gte: new Date(`${year}-01-01`),
          $lt: new Date(`${year + 1}-01-01`),
        },
      },
    },
    {
      $group: {
        _id: { month: { $month: "$createdAt" } },
        revenue: { $sum: "$amount" },
        order_count: { $sum: 1 },
      },
    },
    { $sort: { "_id.month": 1 } },
    {
      $project: {
        _id: 0,
        year: { $literal: year },
        month: "$_id.month",
        revenue: { $round: ["$revenue", 2] },
        order_count: 1,
      },
    },
  ]);
}

Python Example

from pymongo import MongoClient
from datetime import datetime

client = MongoClient("mongodb://admin:password@localhost:27017/")
db = client["myapp"]

def get_monthly_revenue(year: int) -> list:
    pipeline = [
        {
            "$match": {
                "status": {"$in": ["delivered", "processing"]},
                "createdAt": {
                    "$gte": datetime(year, 1, 1),
                    "$lt": datetime(year + 1, 1, 1),
                },
            }
        },
        {
            "$group": {
                "_id": {"month": {"$month": "$created_at"}},
                "revenue": {"$sum": "$amount"},
                "order_count": {"$sum": 1},
            }
        },
        {"$sort": {"_id.month": 1}},
        {
            "$project": {
                "_id": 0,
                "month": "$_id.month",
                "revenue": {"$round": ["$revenue", 2]},
                "order_count": 1,
            }
        },
    ]
    return list(db.orders.aggregate(pipeline))

What’s Next?

You can now transform and analyze data with the aggregation pipeline. Next: MongoDB indexing — making your queries fast.

Next: Database Tutorial #12: MongoDB Indexing and Performance