MongoDB aggregation pipeline 2026

Пошаговое руководство по использованию MongoDB Aggregation Pipeline: от базового $match/$group до $lookup, $facet и оптимизаций. Примерное время выполнения: 45–90 минут.

К концу руководства вы сможете собрать рабочие aggregation pipeline для типичных аналитических задач и оптимизировать их под MongoDB 7.2. Время выполнения полного руководства: примерно 45–90 минут в зависимости от практики.

Когда pipeline удобнее SQL?

Aggregation pipeline удобнее SQL в сценариях, где данные денормализованы, требуется потоковая трансформация документов или многослойная агрегация на стороне БД без лишних round-trip к приложению. В 2025–2026 годах команды всё чаще используют pipeline для временных рядов, событийной аналитики и ETL-процессов, когда нужно выполнить несколько последовательных преобразований одного документа: фильтрацию, проекции, вычисления, джойны и фасеты. Типичный пример — подсчёт сессионной аналитики по событиям клиента, где один документ содержит массив событий.

Преимущества pipeline перед SQL: встроенные этапы (например, $setWindowFields для оконных функций, появившиеся и стабилизированные к 2025), возможность работать с вложенными массивами без JOIN таблиц и гибкость при изменении схемы. Недостаток — сложность отладки длинных конвейеров и чувствительность к объёму RAM при группировках и фасетах.

Шаг 1: $match и $group

Задача: посчитать количество событий по типу и по пользователю за последний месяц. Используем коллекцию events, документы формата {userId, type, ts}.

Команда в mongosh 1.11 (выполнение ~10–50 ms локально на тестовой выборке 100k):

db.events.aggregate([
  { $match: { ts: { $gte: new Date(new Date()-1000*60*60*24*30) } } },
  { $group: { _id: { userId: "$userId", type: "$type" }, count: { $sum: 1 } } },
  { $sort: { "count": -1 } }
])

Пояснение: $match должен быть первым для использования индексного сканирования по полю ts. $group агрегирует по составному ключу.

Ожидаемый вывод (пример):

{ "_id" : { "userId" : "user123", "type" : "click" }, "count" : 542 }
{ "_id" : { "userId" : "user456", "type" : "view" }, "count" : 431 }

Типичная ошибка: отсутствие индекса по ts приводит к полному скану коллекции и OOM при больших данных.

Ошибка: planExecutor error during aggregation :: caused by :: Cannot allocate memory

Фикс: создать индекс и уменьшить рабочую выборку или использовать $limit и периодизацию.

db.events.createIndex({ ts: 1 })
// Создание индекса: ~12s на 1M документов на SSD

Если у вас ограничено RAM, добавьте $project перед $group, чтобы сократить поля, и используйте опцию allowDiskUse: true:

db.events.aggregate(pipeline, { allowDiskUse: true })

Ожидаемое поведение: aggregation сохранит промежуточные данные на диск; производительность упадёт, но OOM уйдёт.

Шаг 2: $lookup

Задача: приписать к каждому событию профиль пользователя из коллекции users по userId. Коллекция users содержит до 500k документов.

db.events.aggregate([
  { $match: { ts: { $gte: ISODate("2026-01-01T00:00:00Z") } } },
  { $lookup: {
      from: "users",
      localField: "userId",
      foreignField: "_id",
      as: "user"
  } },
  { $unwind: "$user" },
  { $project: { user: { name: 1, email: 1 }, type: 1, ts: 1 } }
])

Пояснение: $lookup выполняет left-join эквивалент; предпочтительно, чтобы users._id был индексирован (обычно это primary key).

Ожидаемый вывод (пример):

{ "_id" : ObjectId("..."), "type" : "purchase", "ts" : ISODate("2026-02-10T12:00:00Z"), "user" : { "name" : "Anna", "email" : "anna@example.com" } }

Типичная ошибка: если users._id не индексирован, lookup выполняет COLLSCAN по users и время запроса растёт экспоненциально.

Проблема: high latency, план показывает $lookup -> unindexed COLLSCAN

Фикс: убедитесь, что внешний ключ индексирован — обычно это _id. Для кастомного поля создайте индекс:

db.users.createIndex({ _id: 1 }) // обычно уже есть
// или для кастомного ключа
db.users.createIndex({ userId: 1 })

Дополнительный приём: если связка много событий -> один пользователь и коллекция users небольшая (<=100MB), можно загрузить users в память приложения и делать join на уровне приложения. См. материалы по кешированию в devops.

Шаг 3: $facet

Задача: одновременно получить 3 результата — топ событий по типу, распределение по часам и общую статистику — в одном запросе. Используем $facet для параллельных веток в pipeline.

db.events.aggregate([
  { $match: { ts: { $gte: new Date(new Date()-1000*60*60*24*7) } } },
  { $facet: {
      topTypes: [ { $group: { _id: "$type", c: { $sum: 1 } } }, { $sort: { c: -1 } }, { $limit: 10 } ],
      hourly: [ { $project: { hour: { $hour: "$ts" } } }, { $group: { _id: "$hour", c: { $sum: 1 } } }, { $sort: { _id: 1 } } ],
      stats: [ { $group: { _id: null, total: { $sum: 1 }, uniqueUsers: { $addToSet: "$userId" } } }, { $project: { total: 1, uniqueCount: { $size: "$uniqueUsers" } } } ]
  } }
])

Пояснение: $facet создаёт независимые под-пайплайны и возвращает результаты в одном документе. Это удобно для dashboard-метрик, но потребляет память: сумма промежуточных наборов может быть значительной.

Ожидаемый вывод (упрощённый):

{
  "topTypes" : [ { "_id" : "click", "c" : 12450 }, ... ],
  "hourly" : [ { "_id" : 0, "c" : 512 }, ... ],
  "stats" : [ { "total" : 543210, "uniqueCount" : 12345 } ]
}

Типичная ошибка: OOM при сборе больших поднаборов в facet.

Error: command failed: command aggregate failed: Exceeded memory limit for $group stage

Фикс: использовать allowDiskUse: true, сократить поля в ранних этапах ($project), или разбить задачу на несколько запросов по времени.

Пример с опцией дисковой поддержки:

db.events.aggregate(pipeline, { allowDiskUse: true })
// Опция может увеличить время выполнения до 2–5x, но снимает ограничение RAM.

Шаг 4: $sort и $limit

Задача: получить топ-20 пользователей по количеству событий. Правильный порядок этапов критичен: сначала $group, затем $sort, затем $limit. Если есть индекс покрытия, стоит использовать его до группировки.

db.events.aggregate([
  { $match: { ts: { $gte: ISODate("2026-03-01T00:00:00Z") } } },
  { $group: { _id: "$userId", events: { $sum: 1 } } },
  { $sort: { events: -1 } },
  { $limit: 20 }
])

Ожидаемый результат:

{ "_id" : "user789", "events" : 12433 }
{ "_id" : "user123", "events" : 11220 }
... 18 строк ...

Типичная ошибка: если выполнить $sort на непроиндексированном большом множестве, MongoDB тратит память на сортировку.

Проблема: Sort exceeded memory limit, consider adding index or increasing RAM

Фикс: использовать индекс по полю, участвующему в сортировке, либо применять allowDiskUse: true и предварительно уменьшать набор через $match или $project. Альтернатива: поддерживать агрегированные счётчики в отдельной коллекции обновляемой через change streams или application layer.

Шаг 5: $project и вычисления полей

Задача: сформировать итоговую проекцию с вычислением стоимости заказа и приведением дат в нужную зону времени. Используем $project и выражения.

db.orders.aggregate([
  { $match: { status: "paid", createdAt: { $gte: ISODate("2026-01-01T00:00:00Z") } } },
  { $project: {
      orderId: 1,
      totalCents: { $round: [ { $multiply: ["$unitPrice", "$quantity"] }, 0 ] },
      createdLocal: { $dateToString: { date: "$createdAt", timezone: "Europe/Moscow", format: "%Y-%m-%d %H:%M:%S" } }
  } },
  { $limit: 50 }
])

Ожидаемый вывод:

{ "orderId" : "ord_001", "totalCents" : 1299, "createdLocal" : "2026-03-10 16:20:05" }

Типичная ошибка: использование операций с нечисловыми типами приводит к NaN или ошибкам типов.

Ошибка: Cannot apply $multiply to types: string and int

Фикс: привести типы заранее: $toDouble или хранить цены в числовом поле. Пример приведения:

{ $project: { unitPrice: { $toDouble: "$unitPrice" } } }

Как оптимизировать?

Оптимизация aggregation pipeline включает индексирование, минимизацию передаваемых полей, использование allowDiskUse осознанно и применение кеширования. Конкретные шаги и рекомендации:

• Индексы: создавайте индекс по полям, использующимся в $match (например, {ts:1}). Индексная селективность важна: индекс по {userId:1, ts:-1} уменьшит нагрузку при запросах по пользователю за период.
• Порядок этапов: $match → $project → $group → $sort/$limit. Это уменьшает объём данных до тяжёлых операций.
• Используйте allowDiskUse:true для больших группировок; ожидаемое увеличение времени: 2–5x, но RAM не будет переполняться.
• Поддерживайте предагрегированные коллекции для тяжёлых, но часто запрашиваемых метрик (например, ежедневные счётчики). Обновления можно делать через change streams и background workers.
• Используйте шардирование для наборов > 100GB и нагрузок с большим параллелизмом. Минимальные требования для кластера шардирования: 3 конфигурации + 3 реплики шардов, суммарно 12+ vCPU и 32+ GB RAM для production на 2026.
• Профилирование: выполняйте explain("executionStats") и анализируйте планы. Пример: db.events.aggregate(pipeline).explain("executionStats"). Ожидаемая информация: docsExamined, totalKeysExamined, executionTimeMillis.

Пример анализа плана (выдержка):

"executionStats" : {
  "executionTimeMillis" : 78,
  "totalKeysExamined" : 1200,
  "totalDocsExamined" : 1500
}

Если totalDocsExamined значительно выше ожидаемого, проверьте отсутствие COLLSCAN и улучшите индекс.

Дополнительно: используйте Redis 7 (2025) как кэш для готовых результатных наборов dashboard-метрик. См. статью в категории database для практик кеширования.

Какие антипаттерны?

Список распространённых антипаттернов при использовании aggregation pipeline и способы их избегать.

• Антипаттерн: длинные монолитные конвейеры (>20 этапов) без промежуточной валидации. Проблема: трудно отлаживать и поддерживать, растёт время выполнения. Решение: разбивать на логические блоки и тестировать промежуточные результаты.
• Антипаттерн: использование $facet для большого объёма данных без allowDiskUse. Последствие: OOM. Решение: разбить запросы, выполнить параллельно или включить дисковую поддержку.
• Антипаттерн: джойны с коллекциями, которые не индексированы по полю связи. Последствие: COLLSCAN и рост latency. Решение: индексировать внешние ключи либо вынести join на уровень ETL/приложения.
• Антипаттерн: хранение больших массивов в одном документе и последующая группировка внутри документа. Последствие: BSON size limit (16MB) и ухудшение обновлений. Решение: нормализовать либо лимитировать размер массивов, использовать bucket pattern.
• Антипаттерн: хранение денег в float. Последствие: погрешности при суммировании. Решение: хранить в целых (cents) и проектировать вычисления в целых.

Если итоговые метрики нужны часто и рассчитаны по одним и тем же ключам — поддерживайте предагрегированные коллекции и обновляйте их инкрементально через change streams.

Частые вопросы

Как выбрать между $lookup и denormalization?

Выбор зависит от характера данных и частоты обновлений. Если связанные данные редко меняются и читаются вместе с основными документами, денормализация уменьшит задержки и уберёт необходимость в join на чтении. Если же отношение «один-ко-многим» с большим количеством связанных записей или частыми изменениями в связанной сущности, выгоднее хранить отдельную коллекцию и использовать $lookup. Для high-read сценариев предпочтительнее денормализация с механизмом обновления данных через батч-процессы или change streams. При выборе учитывайте размер документов: MongoDB ограничивает BSON до 16MB, поэтому нельзя релизовать полную денормализацию для очень больших связанных массивов.

Что делать, если $group вызывает OOM?

Первое действие — включить allowDiskUse: true для переноса промежуточных массивов на диск. Второй шаг — сократить объём данных перед $group с помощью $match и $project. Третье — разбить агрегацию по временным интервалам и затем объединить результаты. В production-инфраструктуре хорошее решение — поддержание предагрегатов (daily/hourly) и шардирование коллекции, если объём превышает 100–200 GB. Наконец, анализируйте план через .explain("executionStats"), чтобы понять, где именно потребление памяти критично.

Почему мой $lookup медленный, хотя users маленькая коллекция?

Даже маленькая коллекция может вызвать задержки, если джойны выполняются миллионами раз без индекса по внешнему полю. Проверьте, индексирован ли foreignField. Дополнительно убедитесь, что размер документов небольш, и что сеть/IO на сервере не перегружены. На тестовой машине Docker образ mongo:7.2 (~550 MB) загружается за ~8–20 секунд на типичном SSD; проблемы с IO напрямую влияют на время $lookup. Если latency остаётся высокой, рассмотрите кеширование связанной коллекции в RAM или на стороне приложения.

Когда стоит применять $setWindowFields?

$setWindowFields полезен для расчёта скользящих средних, ранжирования и других оконных функций без необходимости клиента аггрегировать данные. Используйте его для аналитики по времени, когда нужна информация о предыдущих значениях (lag/lead) или округлённые скользящие метрики. Поскольку это относительно тяжёлая операция, убедитесь, что входной набор предварительно отфильтрован и что вы ограничиваете объем обработанных документов — например, применяйте $match по диапазону времени.

Сколько RAM нужно для фасетов и больших группировок?

Минимальная рекомендация для серьёзных фасетов: 8 GB RAM на ноду при обработке десятков миллионов документов; для production-аналитики — 16–32 GB RAM и SSD NVMe. Если вы планируете использовать sharded cluster для аналитики, общий объём RAM и IO должен масштабироваться пропорционально объёму данных и требуемой параллельности. Всегда тестируйте с реалистичными данными: на локальной машине с 8 GB одна фасет-операция по 10M документов может завершаться за минуты с включённым дисковым режимом.

Дополнительные материалы и практические примеры по оптимизации агрегатов находятся в рубриках database и devops.