Aggregation pipeline - это массив стадий. Каждая стадия получает документы, которые выдала предыдущая, меняет их и передаёт дальше. $match фильтрует, $group сворачивает много документов в один на каждый ключ, $lookup присоединяет другую коллекцию, а порядок, в котором вы их расставили, определяет, будет ли запрос использовать индекс, сколько памяти ему нужно и закончится ли он за 40 миллисекунд или упадёт с ошибкой памяти. Коротко: сначала фильтруйте, потом группируйте, соединяйте в последнюю очередь и смотрите в explain(), прежде чем доверять любому pipeline на коллекции, которая будет расти.
В этом руководстве мы шаг за шагом собираем один pipeline на небольшом наборе данных магазина, объясняем, что делает каждая распространённая стадия, показываем, какие стадии могут использовать индекс, а какие нет, и разбираем лимит в 100 МБ, о который спотыкаются, когда отчёт впервые запускается на реальных данных. Предполагается, что вы уже умеете подключаться и выполнять запросы; если нет, начните с строк подключения MongoDB для начинающих.
Что такое pipeline и чем он отличается от find()#
find() возвращает документы в том виде, в каком они хранятся, при желании отфильтрованные и урезанные по полям. aggregate() может вернуть то, чего никогда не было в хранилище: суммы по клиентам, количество заказов по дням, заказы вместе с клиентом, который их сделал. Всё, что в SQL вы написали бы через GROUP BY, JOIN или оконную функцию, в MongoDB проходит через pipeline.
Pipeline запускается так: db.collection.aggregate([ stage1, stage2, ... ]). Документы проходят через стадии по порядку, и каждая стадия видит только то, что выдала предыдущая. Из этого следуют два вывода, и к одному из них сводится большинство проблем с pipeline:
- Стадия может использовать только те поля, которые к этому моменту ещё существуют. Если ранний
$projectили$groupубрал поле, более поздний$matchпо нему ничего не найдёт, и никакая ошибка об этом не сообщит. - Стадии, которым нужен весь вход целиком, прежде чем они что-то выдадут (
$group,$sortбез индекса,$bucket), держат документы в памяти. Стадии, пропускающие документы по одному ($match,$project,$addFields,$limit), почти ничего не стоят.
В примерах ниже две коллекции. В orders один документ на заказ, в customers один на аккаунт:
db.orders.insertMany([ { customerId: 1, status: "paid", total: 42.5, createdAt: ISODate("2026-10-01T09:12:00Z"), items: [{ sku: "A-100", qty: 2, price: 10 }, { sku: "B-200", qty: 1, price: 22.5 }] }, { customerId: 2, status: "paid", total: 15, createdAt: ISODate("2026-10-02T14:40:00Z"), items: [{ sku: "A-100", qty: 1, price: 10 }, { sku: "C-300", qty: 1, price: 5 }] }, { customerId: 1, status: "refunded", total: 10, createdAt: ISODate("2026-10-03T08:05:00Z"), items: [{ sku: "A-100", qty: 1, price: 10 }] }])db.customers.insertMany([ { _id: 1, name: "Nino", plan: "pro" }, { _id: 2, name: "Dato", plan: "free" }])Выполняйте их в mongosh. Те же документы стадий без изменений работают из драйвера; в Node.js, Python и Go pipeline - это просто массив объектов.
$match: фильтруйте первым и используйте индекс#
$match принимает тот же синтаксис запроса, что и find(). Это самая важная стадия pipeline, потому что только она может использовать индекс и только она может удешевить всё, что идёт после неё.
db.orders.aggregate([ { $match: { status: "paid", createdAt: { $gte: ISODate("2026-10-01") } } }])$match в самом начале pipeline планируется так же, как фильтр в find(): если индекс покрывает status и createdAt, MongoDB просматривает диапазон индекса, а не всю коллекцию. $match дальше по pipeline работает с результатом предыдущих стадий, поэтому индекс коллекции использовать не может вообще. Отсюда правило "match как можно раньше".
В одном случае оптимизатор помогает. Если вы пишете $match после $project или $addFields, а условие затрагивает только поля, которые они не меняли, MongoDB сам переносит $match вперёд. Если же условие зависит от поля, вычисленного предыдущей стадией, перенести его нельзя, и полагаться на это не стоит: пишите pipeline в том порядке, в котором хотите его выполнить.
Две детали, о которые обжигаются:
- Обычный
$matchсравнивает поле со значением, а не с другим полем. Пути к полям вроде"$total"читаются только в выражениях. Используйте$expr, когда нужно сравнить два поля одного документа:{ $match: { $expr: { $gt: ["$total", "$limit"] } } }. В общем случае$matchс$exprне может использовать индекс для такого сравнения, так что применяйте его осознанно. - Даты сравниваются как даты, только если хранятся как даты. Если
createdAtвставили строкой ISO, условие$gte: ISODate(...)ничего не найдёт. Выполнитеdb.orders.findOne()и посмотрите на тип, прежде чем писать диапазон.
$group: один документ на ключ#
$group сворачивает все входящие документы с одинаковым выражением _id в один выходной документ и считает по группе аккумуляторы. Выручка по клиентам из оплаченных заказов:
db.orders.aggregate([ { $match: { status: "paid" } }, { $group: { _id: "$customerId", orders: { $sum: 1 }, revenue: { $sum: "$total" }, average: { $avg: "$total" }, lastOrder: { $max: "$createdAt" } } }, { $sort: { revenue: -1 } }])_id группы - это то, по чему вы группируете. Путь к полю ("$customerId"), целый вложенный документ для составного ключа ({ customer: "$customerId", day: { $dateToString: { format: "%Y-%m-%d", date: "$createdAt" } } }) или null, чтобы свернуть весь вход в один документ, так считают общий итог.
Вот аккумуляторы, которые понадобятся чаще всего:
| Аккумулятор | Результат |
|---|---|
$sum | Сумма поля или 1 на документ для подсчёта |
$avg | Среднее значение поля, отсутствующие значения игнорируются |
$min, $max | Наименьшее или наибольшее значение в группе |
$first, $last | Значение из первого или последнего документа (имеет смысл только после $sort) |
$push | Массив всех значений, дубликаты сохраняются |
$addToSet | Массив уникальных значений |
$push и $addToSet опаснее остальных. Они держат в памяти каждое значение каждой группы, а готовая группа - это один документ с ограничением размера 16 МБ. $group с _id: "$customerId" и $push: "$$ROOT" на большой коллекции - стандартный способ упереться сразу в оба лимита. Если вам нужны только три последних заказа каждого клиента, используйте аккумуляторы top-N ($topN, $firstN) в версии, где они есть, а не складывайте всё в массив и не обрезайте его потом.
Отчёт по датам - та же стадия с другим ключом. Заказы по дням:
db.orders.aggregate([ { $match: { createdAt: { $gte: ISODate("2026-10-01"), $lt: ISODate("2026-11-01") } } }, { $group: { _id: { $dateToString: { format: "%Y-%m-%d", date: "$createdAt" } }, orders: { $sum: 1 } } }, { $sort: { _id: 1 } }])Ключ - строка в UTC, если не передать timezone. Если отчёты нужны людям из одного часового пояса, передайте его (timezone: "Asia/Tbilisi"), иначе заказ, сделанный в 01:00 по местному времени, попадёт в предыдущий день.
$project, $addFields, $unwind и $count#
Четыре небольшие стадии делают большую часть переформатирования вокруг $match и $group.
$project оставляет, убирает или вычисляет поля. { $project: { _id: 0, customerId: 1, total: 1 } } вернёт только эти два поля. $addFields (и его псевдоним $set) добавляет вычисляемые поля, сохраняя все остальные, что обычно и нужно в середине pipeline.
$unwind превращает поле-массив в по одному документу на элемент. Так вы агрегируете по items:
db.orders.aggregate([ { $match: { status: "paid" } }, { $unwind: "$items" }, { $group: { _id: "$items.sku", units: { $sum: "$items.qty" }, revenue: { $sum: { $multiply: ["$items.qty", "$items.price"] } } } }, { $sort: { revenue: -1 } }, { $limit: 5 }])$unwind размножает документы, поэтому ставьте его после $match и следите за разрастанием: коллекция из миллиона заказов по десять позиций превращается в десять миллионов документов, проходящих через следующую стадию. Документы, у которых массива нет или он пуст, отбрасываются, если не написать { $unwind: { path: "$items", preserveNullAndEmptyArrays: true } }.
$count - это стадия, которая выдаёт один документ с числом входных документов: { $count: "paidOrders" }. $sort, $limit и $skip работают как в find(), с одним отличием, которое важно дальше: $sort, за которым сразу идёт $limit, оптимизируется в сортировку top-k, где в памяти держатся только k документов.
$lookup: соединение коллекций#
$lookup - это left outer join в MongoDB. Для каждого входящего документа он достаёт подходящие документы из другой коллекции и прикрепляет их массивом:
db.orders.aggregate([ { $match: { status: "paid" } }, { $lookup: { from: "customers", localField: "customerId", foreignField: "_id", as: "customer" } }, { $unwind: "$customer" }, { $project: { total: 1, "customer.name": 1, "customer.plan": 1 } }])as всегда массив, даже когда совпал ровно один документ, поэтому дальше идёт $unwind. Без него вы получите customer: [ { ... } ], что один раз удивляет почти всех. Если ни один документ не совпал, массив пуст, и $unwind выше отбросит такой заказ. Добавьте preserveNullAndEmptyArrays: true, если заказ должен пережить отсутствующего клиента.
Три правила держат $lookup быстрым:
- Индексируйте внешнее поле. Для каждого входящего документа
$lookupвыполняет запрос к другой коллекции. По_idиндекс уже есть. ПоcustomerIdвorders(когда вы соединяете в обратную сторону) его нет, и без индекса каждый поиск сканирует всю коллекцию: тысяча входящих документов против ста тысяч внешних - это сто миллионов сравнений. - Сначала сократите вход. Каждый документ, дошедший до
$lookup, стоит одного запроса.$matchи$limitставьте до него, никогда не после. - Соединяйте в направлении от малой стороны. Начать с нескольких документов, переживших фильтр, и искать по индексу в большой коллекции дёшево. Наоборот - нет.
Когда нужно отфильтровать или переформатировать присоединяемые документы, используйте форму с pipeline, передавая локальные поля через let:
db.customers.aggregate([ { $match: { plan: "pro" } }, { $lookup: { from: "orders", let: { cid: "$_id" }, pipeline: [ { $match: { $expr: { $eq: ["$customerId", "$$cid"] }, status: "paid" } }, { $project: { _id: 0, total: 1, createdAt: 1 } }, { $sort: { createdAt: -1 } }, { $limit: 3 } ], as: "recentOrders" } }])Использование индексов внутри $lookup в форме с pipeline ограниченнее, чем в простой форме. Равенство через $expr по полю может использовать индекс, но проверьте это через explain(), а не предполагайте; об этом ниже. Для соединения, которое вы выполняете постоянно, ответ часто не в более быстром $lookup, а в другой схеме: встраивайте то, что всегда читается вместе, как описано в статье схема и индексы MongoDB. $lookup - хороший инструмент для отчётов и дорогой для горячего пути веб-запроса.
Присоединённый массив тоже лежит внутри одного документа, поэтому на него действует лимит документа в 16 МБ. Если присоединять к клиенту все когда-либо сделанные им заказы без $limit, рано или поздно запрос упадёт на клиенте с самым большим числом заказов.
Какие стадии используют индекс#
Индекса могут коснуться только стадии в начале pipeline, и не все:
| Стадия | Использование индекса |
|---|---|
$match в начале | Да, как в find() |
$sort в начале или сразу после $match с индексом | Да, если порядок даёт индекс |
$group | Нет, но $sort с $first или $last может использовать индекс по ключу сортировки |
$lookup | Поиски во внешней коллекции используют её индексы |
$project, $addFields, $unwind | Нет - они преобразуют документы |
Любая стадия после $group или $project | Нет - это уже не документы коллекции |
Логика простая: индекс - это структура над хранимыми документами. Как только стадия меняет форму потока, указывать индексу не на что. Поэтому порядок "match, sort, group, lookup" - порядок по умолчанию, а pipeline, который начинается с $project или $addFields и только потом фильтрует, - тревожный знак.
Для pipeline, который фильтрует по равенству и сортирует, действуют те же правила составных индексов, что и для find(): сначала поля равенства, затем сортировка, затем диапазон. Как упорядочить поля, объясняет разбор правила ESR. Для отчёта по оплаченным заказам с датами выше:
db.orders.createIndex({ status: 1, createdAt: -1 })обслуживает и $match по status, и диапазон по createdAt, и отдаёт документы уже отсортированными по дате, если pipeline этого требует.
Как читать explain() для pipeline
Не гадайте, использовался ли индекс. Спросите:
db.orders.explain("executionStats").aggregate([ { $match: { status: "paid", createdAt: { $gte: ISODate("2026-10-01") } } }, { $group: { _id: "$customerId", revenue: { $sum: "$total" } } }])В выводе найдите входную стадию. IXSCAN с именем индекса означает, что индекс использован. COLLSCAN означает, что прочитан каждый документ. Сравните totalDocsExamined и nReturned на первой стадии: 500 000 просмотренных документов ради 800 возвращённых значат, что фильтр не селективен для имеющегося индекса или индекса нет. В pipeline с $lookup вывод explain показывает доступ к внешней коллекции отдельно, и COLLSCAN там - это недостающий индекс по foreignField.
Если первая стадия читается как $cursor с блоком queryPlanner, это та часть, которая передана движку запросов, и только её может обслужить индекс. Всё, что перечислено после, выполнялось уже в слое агрегации.
Простой explain() (без уровня детализации) только строит план и никогда не выполняет pipeline, это безопасный выбор для боевой базы. executionStats pipeline выполняет, поэтому на большой коллекции запускайте его на копии или с $limit, который в обычной работе вам не нужен.
Лимит 100 МБ и allowDiskUse#
Каждая блокирующая стадия pipeline может использовать до 100 мегабайт оперативной памяти. Блокирующие стадии - те, которым нужен весь вход, прежде чем выдать результат: $group, $sort (когда порядок не даёт индекс), $bucket, $bucketAuto, $setWindowFields на больших партициях и некоторые другие. Стадии, которые пропускают документы потоком, такие как $match, $project и $addFields, в лимит не входят. Лимит действует на каждую стадию, а не на весь pipeline, поэтому pipeline с $group и следующим за ним $sort может использовать по 100 МБ на каждую.
Когда стадия превышает его, поведение зависит от версии сервера и от опции allowDiskUse:
- В MongoDB 6.0 и новее серверная настройка
allowDiskUseByDefaultравнаtrue, если развёртывание её не отключило, поэтому стадия, переросшая 100 МБ, пишет временные файлы на диск и продолжает работу, но медленнее. - В более ранних версиях или там, где
allowDiskUseByDefaultравнаfalse, стадия падает с ошибкой вродеSort exceeded memory limit of 104857600 bytes, but did not opt in to external sortingилиExceeded memory limit for $group, but didn't allow external sort. ПередайтеallowDiskUse: trueв опциях, чтобы разрешить сброс на диск.
db.orders.aggregate( [ { $group: { _id: "$customerId", orders: { $push: "$_id" } } }, { $sort: { _id: 1 } } ], { allowDiskUse: true })В драйвере та же опция передаётся в аргументе options метода aggregate(). Точное написание смотрите в документации вашего драйвера, оно следует соглашениям языка.
allowDiskUse - это страховочная сетка, а не решение. Он превращает жёсткий отказ в медленный запрос: стадия пишет отсортированные куски в каталог _tmp внутри пути к базе и затем сливает их, что стоит дискового ввода-вывода и на небольшом тарифе конкурирует со всем остальным, что делает сервер. Ему также нужно свободное место в каталоге данных. Pipeline, который регулярно сбрасывает данные на диск, о чём-то вам говорит, и обычные средства по порядку такие:
- Перенесите
$matchраньше или добавьте нужный ему индекс, чтобы до блокирующей стадии доходило меньше документов. - Уберите ненужные поля через
$projectперед$groupили$sort. Стадия держит документы целиком, и документ в 4 КБ с нужным вам полем в 40 байт обходится в сто раз дороже, чем должен. - Поставьте
$limitсразу после$sort, чтобы сохранялись только первые несколько. - Замените
$push: "$$ROOT"и подобное небольшим аккумулятором, который хранит одно число. - Для отчёта, который вы запускаете ежедневно по миллионам документов, предвычислите его в сводную коллекцию через
$merge(или$out) по расписанию и читайте сводку.
Два ограничения не связаны с цифрой 100 МБ, и allowDiskUse на них не влияет. Один документ не может превышать 16 МБ, будь то сохранённый документ, слишком выросшая группа или результат $lookup. А $facet, который запускает несколько подпайплайнов на одном и том же входе, держит свой вывод в одном документе, так что весь его результат должен уместиться в 16 МБ.
Лимит 100 МБ также не означает, что серверу нужно 100 МБ свободной памяти на каждую стадию: это потолок, а небольшой $group использует несколько килобайт. Но на тарифе с 1 ГБ памяти несколько пользователей, одновременно запускающих крупные отчёты, могут занять больше, чем есть у кэша WiredTiger и операционной системы. Раздел о размерах в статье схема и индексы MongoDB объясняет, как кэш и остальной процесс делят память.
Разбор pipeline: от медленного к быстрому#
Возьмём месячный отчёт о выручке с именем и тарифом клиента, написанный так, как его проще всего написать:
db.orders.aggregate([ { $lookup: { from: "customers", localField: "customerId", foreignField: "_id", as: "customer" } }, { $unwind: "$customer" }, { $match: { status: "paid", createdAt: { $gte: ISODate("2026-10-01") } } }, { $group: { _id: "$customer.plan", revenue: { $sum: "$total" }, orders: { $sum: 1 } } }, { $sort: { revenue: -1 } }])Он возвращает верный ответ и неправ тремя способами. $match стоит после $lookup, поэтому соединение выполняется для каждого заказа за всё время. Поля в $match принадлежат заказам, так что оптимизатор часто может передвинуть его вперёд, но такую перестановку лучше сделать самому. А $lookup копирует весь документ клиента в каждый заказ перед $group, которому нужен только plan.
Переписанный вариант:
db.orders.aggregate([ { $match: { status: "paid", createdAt: { $gte: ISODate("2026-10-01") } } }, { $group: { _id: "$customerId", revenue: { $sum: "$total" }, orders: { $sum: 1 } } }, { $lookup: { from: "customers", localField: "_id", foreignField: "_id", as: "customer" } }, { $unwind: "$customer" }, { $group: { _id: "$customer.plan", revenue: { $sum: "$revenue" }, orders: { $sum: "$orders" } } }, { $sort: { revenue: -1 } }])Фильтр использует индекс { status: 1, createdAt: -1 }. Первый $group сводит поток к одному документу на платящего клиента до соединения, так что $lookup выполняется один раз на клиента, а не на заказ. Второй $group суммирует их по тарифам. Выполните explain("executionStats") для обоих вариантов, и разница будет видна в totalDocsExamined и во времени каждой стадии.
Безопасный запуск агрегаций на живой базе#
Несколько привычек не дадут отчёту уронить вместе с собой сайт.
- Сначала пробуйте на ограниченном входе. Пока разрабатываете, ставьте
{ $limit: 1000 }сразу после$matchи убирайте, когда результат верный. - Задайте лимит времени.
maxTimeMSв команде остановит убежавший pipeline:db.orders.aggregate([...], { maxTimeMS: 20000 }). Отчёт, который не укладывается в двадцать секунд, не должен быть веб-запросом. - Находите и останавливайте то, что выполняется.
db.currentOp()показывает операции, аdb.killOp(opid)останавливает одну из них, как описано в разделе о профайлере в статье схема и индексы MongoDB. - Заведите отдельного пользователя для отчётов. Дайте ему роль
readна одну базу, чтобы опечатка в pipeline не добралась до$outили$mergeв боевую коллекцию. Какую роль выбрать, рассказывает статья пользователи и роли MongoDB. - Делайте backup перед любым `$out` или `$merge`.
$outзаменяет целевую коллекцию, когда pipeline заканчивается. Сначала сделайте mongodump, если целевая коллекция вам дорога.
На RE:NODE к тарифу MongoDB прилагаются сгенерированные для вас учётные данные и слоты backup, которые можно запускать вручную или по расписанию на вкладке Schedules, - это разумный способ сделать backup перед pipeline, который пишет данные. Память тарифа - жёсткий лимит: сервер, у которого она закончилась, останавливается и чисто перезапускается, а не уходит в swap, так что очень большой $group в памяти на маленьком тарифе - повод сначала переосмыслить pipeline, а не повышать тариф. Тарифы перечислены на странице хостинга баз данных MongoDB.
Устранение неполадок#
`Exceeded memory limit for $group, but didn't allow external sort.` $group перешёл за 100 МБ. Передайте { allowDiskUse: true }, а затем сделайте настоящее исправление: фильтруйте раньше, убирайте неиспользуемые поля через project, не складывайте через $push документы целиком.
`Sort exceeded memory limit of 104857600 bytes`. Тот же лимит для $sort, за которым нет индекса. Добавьте индекс, соответствующий сортировке, или поставьте после неё $limit.
`BSONObj size: ... is invalid` или "document exceeds maximum size 16777216". Группа или массив из $lookup стали больше 16 МБ. Ограничьте то, что собираете, соединяйте более узким pipeline или перестройте данные.
Pipeline ничего не возвращает и не выдаёт ошибки. Обычно поля уже нет на этой стадии, $unwind применён к отсутствующему полю или дата хранится строкой. Сократите pipeline до первой стадии, проверьте вывод и добавляйте стадии по одной.
`$lookup` очень медленный. Выполните explain("executionStats") и поищите COLLSCAN во внешней коллекции. Проиндексируйте foreignField и сократите число входящих документов.
`$match` медленный, несмотря на индекс. Вероятно, это не первая стадия, или в фильтре несовпадение типов (число против строки), или он начинается с отрицания либо регулярного выражения с ведущим подстановочным знаком. Посмотрите границы IXSCAN в explain().
Результаты различаются между запусками. $group не возвращает группы в каком-либо определённом порядке, а $first и $last зависят от порядка входа. Если результат зависит от порядка, поставьте $sort раньше.
FAQ#
Быстрее ли aggregation pipeline, чем find()?
Сам по себе нет. Для простого фильтра find() и pipeline только с $match выполняют одну и ту же работу. Pipeline быстрее, чем та же задача в коде приложения, потому что группировка и соединение происходят внутри сервера, а по сети идёт только результат.
Использует ли $match индекс?
Да, когда это первая стадия или перед ним только стадии, через которые оптимизатор может его перенести. $match после $group, $project с переименованием полей или $unwind работает с преобразованными документами и не может использовать индекс коллекции.
Что делает allowDiskUse?
Позволяет блокирующей стадии писать временные файлы на диск при превышении лимита памяти 100 МБ, вместо того чтобы падать. Запрос завершается, но медленнее. В MongoDB 6.0 и новее серверное значение по умолчанию уже разрешает это, если администратор не изменил allowDiskUseByDefault.
Можно ли соединить больше двух коллекций?
Да, цепочкой стадий $lookup, но каждая добавляет запрос на каждый входящий документ, так что стоимость умножается. Если в горячем запросе таких стадий больше двух-трёх, данным, скорее всего, место в одном документе.
Чем $lookup отличается от SQL JOIN?
Это left outer join, который возвращает совпадения массивом внутри каждого документа, а не дополнительными строками. Он выполняет один поиск на каждый входящий документ и плохо подходит для полного соединения двух очень больших коллекций. SQL-базы построены именно для этого; если ваши данные в основном реляционные, сначала прочитайте PostgreSQL или MongoDB.
Когда лучше предвычислять, а не агрегировать?
Когда один и тот же дорогой отчёт читают намного чаще, чем меняются данные. Запускайте pipeline по расписанию с $merge в сводную коллекцию, а приложение пусть читает её. Дашборд, который при каждой загрузке страницы заново агрегирует миллион документов, - обычная ошибка.
