Сложные выборки в MongoDB из конвейеров агрегации от $match до оконных функций

Отдел продаж просит показать выручку по городам за квартал, топ позиций чека и средний чек по месяцам, причём всё это в одном отчёте и с ранжированием внутри каждого города. Обычный метод find с такой задачей не справится: он умеет фильтровать и отдавать документы, но не умеет ни группировать, ни считать суммы, ни ранжировать. В MongoDB для этого существует конвейер агрегации, язык этапов, через который документ проходит как деталь по ленте конвейера, обрабатываясь шаг за шагом. Ниже разобран путь от первого фильтра до оконных функций и сохранения отчёта, с примерами команд, разбором планов и лимитов памяти.

Конвейер агрегации как конструктор из последовательных этапов

Идея проста: в метод aggregate передаётся массив этапов, и каждый этап забирает документы у предыдущего, преобразует и передаёт дальше. Первым делом стоит прореживать поток, поэтому фильтр ставится как можно раньше, а сортировка и срезы под конец, когда строк уже мало. Порядок этапов решает всё: один и тот же набор этапов в другой последовательности даст либо мгновенный отчёт, либо многочасовую перекладку всей коллекции.

Простейший конвейер выглядит так:

db.orders.aggregate([
  { $match: { status: "paid" } },
  { $group: {
      _id: "$city",
      orders: { $sum: 1 },
      revenue: { $sum: "$total" }
  }},
  { $sort: { revenue: -1 } },
  { $limit: 10 }
])

Сначала остаются только оплаченные заказы, затем они сворачиваются по городам, результат сортируется по убыванию выручки и обрезается до десяти строк. Этапов в арсенале несколько десятков, и внутри этапов работают выражения: операторы подсчёта вроде $sum и $avg, операторы массивов, дат и арифметики. Полезно помнить, что конвейер это не SQL: он процедурен по духу, читается сверху вниз как рецепт, и именно поэтому его легко отлаживать по шагам, добавляя этапы по одному и глядя на промежуточный результат в mongosh или в графической утилите Compass.

Фильтрация и группировка через $match $group и $project на реальном отчёте

Пусть заказ хранит дату, город, сумму и статус. Отчёт просит выручку по месяцам и городам сразу, и для этого группировка строится по составному ключу:

db.orders.aggregate([
  { $match: {
      status: "paid",
      created: { $gte: ISODate("2026-01-01") }
  }},
  { $group: {
      _id: {
        month: { $month: "$created" },
        city: "$city"
      },
      revenue: { $sum: "$total" },
      orders: { $sum: 1 },
      avgCheck: { $avg: "$total" }
  }},
  { $sort: { "_id.month": 1, revenue: -1 } }
])

Оператор $sum со значением 1 просто считает документы, а с путём до поля складывает значения. Ключ _id группировки разрешён составной, и обращение к его частям на следующем этапе идёт через точку. Дальше форму ответа правят этапы $project и $set: первый вырезает поля, второй добавляет вычисляемые. Долю города в общей выручке удобнее считать после группировки вторым проходом, где общая сумма собирается отдельной группировкой без ключа, а затем соединяется с разбивкой. Для аккуратных процентов хватает операторов $divide, $multiply и $round:

{ $set: {
    sharePercent: {
      $round: [
        { $multiply: [{ $divide: ["$revenue", "$grandTotal"] }, 100] }, 1
      ]
    }
} }

Здесь есть тонкость, на которой спотыкаются многие: этап $group съедает исходные документы и отдаёт только ключ и накопленные поля. Всё, что нужно для ответа, надо накопить внутри группировки, добавить через $push в массив или досчитать позже через $lookup к самому себе. Решение "сначала сгруппирую, а потом вспомню про нужное поле" в конвейере не работает.

Соединение коллекций через $lookup и разворот массивов $unwind

В реальной схеме заказ ссылается на товары массивом строк с количеством и ценой, а справочник лежит отдельно. Суммарные продажи позиций считаются разворотом массива: этап $unwind размножает документ по элементам, и каждый элемент становится строкой, по которой уже можно группировать:

db.orders.aggregate([
  { $match: { status: "paid" } },
  { $unwind: "$items" },
  { $group: {
      _id: "$items.sku",
      sold: { $sum: "$items.qty" },
      revenue: { $sum: { $multiply: ["$items.qty", "$items.price"] } }
  }},
  { $sort: { sold: -1 } },
  { $lookup: {
      from: "products",
      localField: "_id",
      foreignField: "sku",
      as: "product"
  }},
  { $unwind: "$product" }
])

Оператор $lookup это местный аналог соединения таблиц: по полю localField он подтягивает документы из другой коллекции в массив as. Результат всегда массив, даже если совпадение одно, поэтому после него обычно стоит ещё один $unwind, а чтобы не терять строки без совпадений, развороту передают флаг preserveNullAndEmptyArrays со значением true. Стоимость $lookup зависит от индекса на внешнем поле: без него на каждую входящую строку выполняется полный просмотр справочника, и отчёт из секунд превращается в минуты.

Для сложных условий есть расширенная форма $lookup с собственным подконвейером, где через let передаются переменные из внешнего документа:

{ $lookup: {
    from: "reviews",
    let: { sku: "$_id" },
    pipeline: [
      { $match: { $expr: { $eq: ["$sku", "$$sku"] } } },
      { $group: { _id: null, rating: { $avg: "$stars" } } }
    ],
    as: "rating"
} }

Так соединяют не по равенству ключей, а по любому выражению, например подтягивают среднюю оценку товара сразу в отчёт продаж. Начиная с версии 5.0 сервер умеет соединять и шардированные коллекции, раньше это ограничение приходилось обходить денормализацией. Если объединение нужно постоянно, честнее хранить копию нужных полей прямо в заказе и обновлять её при изменении справочника: одна запись при правке дешевле тысячи соединений при каждом отчёте.

Оконные функции $setWindowFields для рангов и нарастающих итогов

Долгое время конвейер не умел главного из арсенала аналитики: посчитать ранг строки среди соседей или нарастающую сумму по порядку. С версии 5.0 появился этап $setWindowFields, работающий с окнами наподобие оконных функций SQL. Сначала строки разбиваются на секции по partitionBy, внутри секции сортируются по sortBy, и уже по этому порядку считаются выходные поля:

db.orders.aggregate([
  { $match: { status: "paid" } },
  { $setWindowFields: {
      partitionBy: "$city",
      sortBy: { created: 1 },
      output: {
        rank: { $rank: {} },
        runningTotal: {
          $sum: "$total",
          window: { documents: ["unbounded", "current"] }
        }
      }
  }}
])

Запись window со значениями unbounded и current означает весь диапазон от начала секции до текущей строки, так поле runningTotal превращается в нарастающий итог по каждому городу. Ранжирование продающих менеджеров или топ заказов клиента теперь не требует выгрузки в приложение: операторы $rank и $denseRank делают всё на сервере. Скользящее среднее по последним семи документам задаётся окном от минус шести до текущей позиции, а для календарных рядов то же окно можно считать по диапазону дат, а не по количеству строк.

Соседняя задача аналитики, заполнение пробелов в рядах данных, решается парой этапов из версии 5.1. Этап $densify достраивает недостающие точки оси, например каждые сутки в диапазоне отчёта, а $fill закрывает образовавшиеся дыры значениями, интерполированными по соседям или продлёнными вперёд:

db.daily_stats.aggregate([
  { $densify: {
      field: "day",
      range: { step: 1, unit: "day" },
      bounds: "full"
  }},
  { $fill: {
      sortBy: { day: 1 },
      output: { orders: { method: "linear" } }
  }}
])

График на дашборде после такой обработки перестаёт обрываться в выходные и превращается в гладкую линию, не требуя внешних скриптов.

Память этапов и лимит сто мегабайт с параметром allowDiskUse

У скорости конвейера есть физический предел, о котором полезно помнить: этап не имеет права занять в оперативной памяти больше ста мегабайт. Лимит защищает сервер от одного отчёта, съевшего всю память, и на тяжёлых сортировках или группировках порог пробивается быстро. Симптом знаком многим: сообщение об ошибке с упоминанием exceeded memory limit и имени этапа. Лечение простое, разрешение на использование диска передаётся вторым аргументом:

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

При таком режиме сортировка и группировка выкладывают часть данных во временные файлы на диске, и отчёт доходит до конца медленнее, зато стабильно. В версиях сервера, начиная с шестой, разрешение на диск включено по умолчанию, и наоборот, при желании его можно отключить, вернув строгий лимит. Дисковая подстраховка не отменяет главную дисциплину: чем меньше документов доходит до дорогого этапа, тем быстрее и предсказуемее весь конвейер. Ранний $match с индексом, срез $limit до сортировки, когда логика отчёта позволяет, и компактная проекция, выкидывающая ненужные поля, экономят больше, чем любые параметры.

Отдельно стоит следить за накоплением в $push и $addToSet: массивы растут без ограничений и легко ломают лимит на больших группах. Если внутри группы ожидаются тысячи значений, безопаснее считать агрегаты на месте, например $sum и $avg, а не тащить сами значения наверх.

Индексы и разбор плана через explain для быстрых выборок

Медленный конвейер почти всегда начинается с плохого первого этапа. Прежде чем оптимизировать, план смотрят глазами:

db.orders.explain("executionStats").aggregate([
  { $match: { status: "paid", created: { $gte: ISODate("2026-01-01") } } },
  { $sort: { created: -1 } },
  { $limit: 20 }
])

В ответе ищут четыре цифры: totalKeysExamined, сколько просмотрено ключей индекса, totalDocsExamined, сколько прочитано самих документов, nReturned, сколько вернулось, и executionTimeMillis, общее время. Идеальная картина, когда ключей просмотрено ненамного больше, чем возвращено, а чтение документов отсутствует вовсе. Слово COLLSCAN в плане означает полный просмотр коллекции и служит сигналом, что фильтр не использует индекс.

Составной индекс под конкретный запрос строится по правилу ESR: сначала поля равенства, затем поля сортировки, затем диапазоны. Под отчёт по оплаченным заказам за период с сортировкой по дате индекс ложится естественным образом:

db.orders.createIndex({ status: 1, created: -1, city: 1 })

Тот же индекс закрывает и сортировку: раз порядок полей совпадает с требуемым, сервер отдаёт строки прямо из дерева без отдельного этапа сортировки. Если конвейер на выходе использует только индексированные поля и выключает _id, срабатывает покрывающий запрос, документы вообще не читаются, и в плане вместо FETCH остаётся чистый IXSCAN. Выражения $expr внутри $match применяют осторожно: они гибче, но не всегда позволяют использовать индекс так же эффективно, как прямое сравнение, поэтому в критичных отчётах условие стараются переписать обычными операторами.

Разрезы дашбордов через $facet $bucket и сохранение через $merge

Классический дашборд просит несколько разрезов сразу: по городам, по месяцам и топ заказов. Гонять конвейер по кругу для каждого блока расточительно, и с 2016 года в арсенале есть этап $facet, который выполняет несколько подконвейеров над одним и тем же входом и отдаёт их результаты полями одного документа:

db.orders.aggregate([
  { $match: { status: "paid" } },
  { $facet: {
      byCity: [
        { $group: { _id: "$city", revenue: { $sum: "$total" } } },
        { $sort: { revenue: -1 } }
      ],
      byMonth: [
        { $group: { _id: { $month: "$created" }, revenue: { $sum: "$total" } } }
      ],
      topChecks: [
        { $sort: { total: -1 } },
        { $limit: 5 }
      ]
  }}
])

Внутри подконвейеров действуют те же лимиты памяти, поэтому входной поток перед фасетом прореживают и держат подконвейеры компактными. Сегментацию по корзине чека делает этап $bucket с явными границами:

{ $bucket: {
    groupBy: "$total",
    boundaries: [0, 1000, 5000, 25000],
    default: "остальные",
    output: { count: { $sum: 1 } }
} }

Готовый отчёт разумно не пересчитывать при каждом открытии страницы. Этап $merge записывает результат в отдельную коллекцию, обновляя совпадающие документы:

{ $merge: {
    into: "monthly_report",
    whenMatched: "replace",
    whenNotMatched: "insert"
} }

Так ночной прогон кладёт данные в компактную таблицу, а интерфейс читает её обычным find за миллисекунды. Порядок сборки такого отчёта обычно складывается сам собой:

1. сформулировать вопрос отчёта и его разрезы, зафиксировать период;
2. собрать конвейер от раннего фильтра и проверить промежуточные результаты по шагам;
3. прогнать explain, добавить индекс по правилу ESR и убрать полный просмотр из плана;
4. вынести независимые разрезы в $facet и заменить тяжёлые группировки ранними срезами;
5. зафиксировать результат через $merge и поставить пересчёт по расписанию на сервере приложений.

Конвейер агрегации изнутри выглядит устрашающе, но по сути это честный конвейер: сырые заказы входят с одной стороны, готовые цифры выходят с другой. Команда, которая научилась читать его сверху вниз и смотреть план перед оптимизацией, перестаёт выгружать коллекции в электронные таблицы и считает отчёты там, где живут данные.