MongoDB Aggregation Pipelines I Actually Use in Production

When find() is not enough

Early in a project I fetched all orders for a merchant, then reduced totals in JavaScript. It worked until the merchant hit 80k orders. Memory spiked, the event loop stalled, and the API returned 504s.

Aggregation moves that work next to the data. You filter early, project only needed fields, and let MongoDB stream results. The mental model is a Unix pipe: each stage transforms the stream for the next.

Aggregation is also how I implement rollups for invoices and tax reports where floating point in JS would scare accountants. Keep money in integer cents through the pipeline.

Pipeline shape that stays fast

My default order is: $match as early as possible, then $project or $addFields, then $group / $sort / $lookup, then final shaping. Putting $match first lets the planner use indexes. Putting $lookup last when possible avoids joining documents you will throw away.

I avoid $lookup inside hot request paths when a denormalized field would do. Lookups are powerful and expensive. For dashboards that run every minute, I often maintain a summary collection updated by the write path instead.

• $match early — use indexed fields
• $project early — drop bulky fields before $group
• $sort after reducing cardinality when you can
• $lookup only for fields you cannot afford to denormalize
• Prefer $facet for multi-metric dashboards in one round trip

Example: merchant revenue by day

This is the report that replaced my Node reduce. Status and merchantId are indexed. I match first, then group by day in UTC, then sort.

const revenueByDay = await db.collection('orders').aggregate([
  {
    $match: {
      merchantId: new ObjectId(merchantId),
      status: { $in: ['paid', 'refunded'] },
      createdAt: { $gte: start, $lt: end },
    },
  },
  {
    $project: {
      day: {
        $dateToString: { format: '%Y-%m-%d', date: '$createdAt', timezone: 'UTC' },
      },
      netCents: {
        $cond: [
          { $eq: ['$status', 'refunded'] },
          { $multiply: ['$totalCents', -1] },
          '$totalCents',
        ],
      },
    },
  },
  {
    $group: {
      _id: '$day',
      revenueCents: { $sum: '$netCents' },
      orderCount: { $sum: 1 },
    },
  },
  { $sort: { _id: 1 } },
]).toArray();

Joining with $lookup without melting the cluster

I use $lookup for admin views where freshness matters more than micro-latency. Always constrain the local side first. A $lookup after an unbounded collection scan is how you discover your working set size the hard way.

Prefer the pipeline form of $lookup when you need to filter the foreign collection. The simple localField/foreignField form is fine for one-to-one id joins.

const withCustomers = await db.collection('orders').aggregate([
  { $match: { merchantId: oid, createdAt: { $gte: start } } },
  { $limit: 100 },
  {
    $lookup: {
      from: 'customers',
      let: { cid: '$customerId' },
      pipeline: [
        { $match: { $expr: { $eq: ['$_id', '$$cid'] } } },
        { $project: { name: 1, email: 1 } },
      ],
      as: 'customer',
    },
  },
  { $unwind: { path: '$customer', preserveNullAndEmptyArrays: true } },
]).toArray();

Indexes and explain()

If the first $match cannot use an index, nothing downstream saves you. I run explain('executionStats') on new pipelines in staging with production-like data volume. Look for COLLSCAN on large collections and high nReturned versus totalDocsExamined ratios that look wrong.

Compound indexes should mirror the equality fields in $match, then range fields. For the revenue query above I use { merchantId: 1, status: 1, createdAt: 1 }.

Partial indexes that match the $match filter of a hot pipeline can shrink index size. Worth it when the collection is large and the query always includes status: 'paid'.

Mistakes that burned me

Allowing clients to pass arbitrary field names into $project or $match is an injection footgun. Build stages in code from allowlists. Never concatenate user strings into aggregation JSON.

Unbounded $group on high-cardinality keys blew memory on a shared Atlas tier. I now cap date ranges in the API and reject ranges larger than 90 days for heavy reports.

Forgetting allowDiskUse on rare huge sorts caused failures after we upgraded Mongo versions and defaults shifted. I set it explicitly on batch jobs, never on interactive requests if I can avoid spill.

Debugging and observability

I log pipeline name, merchant id, duration, and docs examined when available. Slow aggregation logs have caught missing indexes faster than any dashboard.

Keep pipelines in named functions, not inline in route handlers. That makes them testable with a fixture database and easier to review in PRs.

Facets for dashboard endpoints

Dashboards often need counts, totals, and a small recent list. Three round trips mean three chances to disagree under concurrent writes. $facet runs parallel subpipelines on the same filtered input.

I still $match before $facet so each branch does not rescan the collection. Inside each branch I keep stages minimal. Facets that each $lookup the world will time out together and take the request with them.

const dashboard = await db.collection('orders').aggregate([
  { $match: { merchantId: oid, createdAt: { $gte: start } } },
  {
    $facet: {
      totals: [
        { $group: { _id: null, revenue: { $sum: '$totalCents' }, n: { $sum: 1 } } },
      ],
      byStatus: [
        { $group: { _id: '$status', n: { $sum: 1 } } },
      ],
      recent: [
        { $sort: { createdAt: -1 } },
        { $limit: 10 },
        { $project: { totalCents: 1, status: 1, createdAt: 1 } },
      ],
    },
  },
]).next();

Change streams are not aggregations—and that is fine

When people ask for “real-time aggregations,” they sometimes mean change streams updating a summary document. That is a different tool. I use aggregations for request/response reports and scheduled jobs. I use change streams or write-time increments when the UI needs live counters.

Mixing them poorly—running a full aggregation on every change—will melt Atlas. Update a summary collection incrementally on write, and use aggregation for historical ranges and audits.

Cursor batches and Node memory

toArray() on a huge aggregation result recreates the original problem—Node holds everything. For exports I stream the cursor and write CSV chunks. Set batchSize thoughtfully; default batches can still be large.

For HTTP APIs I paginate with $skip/$limit only on small offsets, or better, range on _id or createdAt. Deep $skip on millions of documents is a known performance trap.

const cursor = db.collection('orders').aggregate(pipeline, { allowDiskUse: true, batchSize: 500 });
for await (const doc of cursor) {
  await writeCsvRow(doc);
}

Testing aggregations without flaky clocks

I seed a fixture database with known ObjectIds and fixed createdAt values. Tests assert exact group keys and sums. Avoid Date.now() in fixtures unless you control the clock.

Snapshot-testing the pipeline array itself helps review in PRs: when someone reorders stages, the diff is obvious. Pair that with an integration test that runs against Mongo Memory Server or a disposable container.

Production incident tip: keep the last slow pipeline in logs as JSON. Being able to paste it into Compass or mongosh cuts mean time to understanding dramatically.

• Stream cursors for large exports; avoid toArray()
• Prefer range pagination over deep $skip
• Fixture dates should be fixed and explicit
• Log slow pipeline JSON for later replay

Putting it into the service layer

I wrap pipelines in repository functions with typed inputs: merchantId, start, end—not free-form stage arrays from the controller. Controllers should not invent aggregation stages from query strings.

Timeouts matter. Set maxTimeMS on heavy reports so a bad plan cannot hold a mongod thread forever. Return a clear 503 or 400 to the client when the range is too wide instead of letting the driver hang until the load balancer cuts the connection.

Finally, revisit denormalization quarterly. If you $lookup the same customer fields on every list view, copying a display name onto the order at write time may be cheaper. Aggregation is powerful; it is not an excuse to skip modeling.

That mix—indexed $match first, careful $group, rare $lookup, faceted dashboards, streamed exports—is the practical aggregation toolkit I actually keep reaching for.

Key takeaways

• Use aggregation when you would otherwise pull large result sets into Node to reduce.
• Put selective $match first so indexes can cut the working set early.
• Shape fields with $project before heavy $group or $lookup stages.
• Verify new pipelines with explain() on realistic data volumes.
• Build stages from allowlists—never interpolate raw user input into pipelines.
• Cap date ranges and cardinality for interactive reports; use batch jobs for huge scans.