Агрегации в MongoDB: $match, $group, $lookup на примерах

Как работает aggregation pipeline

Aggregation pipeline — это массив стадий. Документы проходят их по очереди: каждая стадия получает поток от предыдущей и отдаёт изменённый поток дальше. Так в MongoDB строят отчёты: выручку по месяцам, топ клиентов, продажи по товарам.

Главное правило: сначала уменьшайте объём данных, потом считайте. Фильтр и сортировка в начале, тяжёлые операции и джойны — в конце, на уже маленьком наборе.

Все примеры ниже — на коллекции orders:

{
  "_id": "...",
  "customerId": "...",
  "status": "paid",
  "createdAt": "2026-01-15T10:00:00Z",
  "total": 420,
  "items": [{ "sku": "A-1", "qty": 2, "price": 150 }, { "sku": "B-7", "qty": 1, "price": 120 }]
}

Основные стадии и их аналоги в SQL

Пример 1: топ-10 клиентов за месяц

db.orders.aggregate([
  { $match: {
      status: "paid",
      createdAt: { $gte: ISODate("2026-01-01"), $lt: ISODate("2026-02-01") }
  } },
  { $group: { _id: "$customerId", orders: { $sum: 1 }, revenue: { $sum: "$total" } } },
  { $sort: { revenue: -1 } },
  { $limit: 10 },
  { $lookup: { from: "customers", localField: "_id", foreignField: "_id", as: "customer" } },
  { $unwind: "$customer" },
  { $project: { _id: 0, name: "$customer.name", orders: 1, revenue: 1 } }
])

Обратите внимание на порядок:

• $match стоит первым и может использовать индекс.
• $lookup выполняется после $limit, то есть только для 10 документов, а не для всех клиентов за месяц.
• $unwind здесь превращает массив customer из $lookup в обычный вложенный объект.

Пример 2: продажи по товарам через $unwind

Позиции заказа лежат в массиве items. Чтобы посчитать продажи по SKU, массив нужно развернуть:

db.orders.aggregate([
  { $match: { status: "paid", createdAt: { $gte: ISODate("2026-01-01") } } },
  { $unwind: "$items" },
  { $group: {
      _id: "$items.sku",
      qty: { $sum: "$items.qty" },
      revenue: { $sum: { $multiply: ["$items.qty", "$items.price"] } }
  } },
  { $sort: { revenue: -1 } }
])

$unwind умножает количество документов: заказ с пятью позициями превращается в пять документов. Поэтому ставьте его после $match, а не до.

Индексы, которые помогают pipeline

• Индекс используют только начальные стадии — $match и $sort в начале pipeline. После $group или $unwind данные уже в памяти, и индексы коллекции не работают.
• Для примеров выше подходит составной индекс { status: 1, createdAt: 1 }: сначала поле с точным совпадением, затем диапазон.
• Для $lookup важен индекс на поле foreignField во второй коллекции. Для _id он есть всегда, для других полей его нужно создать, иначе каждый джойн будет сканировать коллекцию.
• Проверяйте план: db.orders.explain("executionStats").aggregate([...]). Если на первой стадии видите COLLSCAN, индекс не используется.

Лимиты памяти

• У стадий, которые держат данные в памяти ($group, $sort), есть лимит памяти на стадию. При превышении запрос либо падает, либо сбрасывает данные на диск — это управляется опцией allowDiskUse. Поведение по умолчанию зависит от версии, поэтому проверьте документацию вашей версии MongoDB.
• Каждый документ результата ограничен 16 МБ. Если $lookup подтягивает тысячи связанных документов в один массив, лимит легко превысить.
• Запись на диск спасает от ошибки, но работает медленнее. Лучшее решение — уменьшить данные через $match и $project до тяжёлых стадий.

Частые ошибки

• $lookup в начале pipeline для всей коллекции — дорогой джойн там, где он не нужен.
• $unwind до фильтрации: объём данных растёт в разы.
• Фильтр по вычисленному полю, которое можно было отфильтровать по исходному полю раньше.
• Тяжёлые отчёты на основной базе в рабочие часы. Для них лучше secondary-реплика или заранее посчитанные данные через $merge в отдельную коллекцию.