Агрегация в MongoDB

Это похоже на агрегатную функцию SQL.

Что такое агрегация?

В программировании мы часто выполняем серию операций над коллекцией элементов. Возьмем следующий пример JavaScript:

let numbers = [{val: 1}, {val: 2}, {val: 3}, {val: 4}];
numbers = numbers
    .map(obj => obj.val) // [1, 2, 3, 4]
    .reduce((prev, curr) => prev + curr, 0) // 10

В этом примере у нас есть две операции, которые выполняются над массивом чисел:

• Во-первых map(): мы берем объекты и преобразуем их в числовые значения.
• Во-вторых reduce(): мы объединяем выходные данные в одно число — сумму чисел.

Операции агрегирования обрабатывают записи данных и возвращают вычисленные результаты.

У нас есть не только возможность агрегировать данные на стороне клиента с помощью JavaScript, но мы можем использовать MongoDB для выполнения операций на сервере над нашими коллекциями, хранящимися в базе данных, прежде чем результат будет возвращен клиенту.

Одноцелевое агрегирование

MongoDB предоставляет два метода агрегирования. Самый простой — одноцелевая агрегация .

Одноцелевые операции агрегирования представляют собой набор вспомогательных методов, применяемых к коллекции для вычисления результата. Эти вспомогательные методы обеспечивают простой доступ к общим процессам агрегирования.

Два из таких методов:

• distinct()
• countDocuments()

Давайте воспользуемся коллекцией с именем «sales», в которой хранятся покупки:

{
    _id: 5bd761dcae323e45a93ccfea,
    saleDate: 2017-06-22T09:54:14.185+00:00,
    items: [
      {
        "name": "printer paper",
        "price": 17.3,
        // ...
      },
    ],
    storeLocation: "Denver",
    customer: {
        age: 40,
        satisfaction: 5,
       // ...
    },
    couponUsed: false,
    purchaseMethod: "In store"
}

Если бы мы хотели определить, какие существуют способы покупки, мы могли бы вызвать distinct() наш скрипт Node.js:

const collection = client.db("sample_supplies").collection("sales");
const distinctPurchaseMethods = await collection.distinct("purchaseMethod");

distinctPurchaseMethods — это массив, содержащий все уникальные методы покупки, хранящиеся в коллекции «sales».

["In store", "Online", "Phone"]

Если бы мы хотели узнать, сколько всего было продано, мы могли бы запустить:

const totalNumberOfSales = await collection.countDocuments();

countDocuments() агрегирует общее количество документов в коллекции и вернет это число для использования. Если нам нужно агрегировать коллекцию на основе одного из вышеперечисленных вспомогательных методов, мы можем использовать одноцелевую агрегацию.

Как использовать MongoDB для агрегирования данных?

Если вам нужно выполнить более сложную агрегацию, вы можете использовать конвейер агрегации MongoDB. Конвейеры агрегации — это последовательности этапов, которые могут запрашивать, фильтровать, изменять и обрабатывать наши документы. Это полная по Тьюрингу реализация, которую можно использовать как (довольно неэффективный) язык программирования.

Прежде чем мы углубимся в код, давайте разберемся, что делает сам конвейер агрегации и как он работает. В конвейере агрегации вы перечисляете серию инструкций на «этапе». Для каждого определенного этапа MongoDB выполняет их один за другим, чтобы предоставить окончательный результат, который вы сможете использовать. Давайте посмотрим на пример использования команды aggregate:

collection.aggregate([
   { $match: { status: "A" } },
   { $group: { _id: "$cust_id", total: { $sum: "$amount" } } }
])

В этом примере мы запускаем этап под названием $match. После запуска этого этапа он передает на него выходные данные $group.

$matchпозволяет нам взять коллекцию элементов и получить только элементы со status значениями A.

После этого мы используем $group для группировки документов по cust_id полю. В рамках этапа $group мы вычисляем сумму всех полей каждого из group них amount.

Помимо $sum, MongoDB предоставляет множество других операторов, которые вы можете использовать в своих агрегатах.

Метод конвейера агрегации

Давайте, например, посмотрим на ту же коллекцию продаж, которую мы использовали ранее. Ниже приведен документ из этой коллекции:

{
  "_id": "5bd761dcae323e45a93ccffb",
  "items": [
    {
      "name": "printer paper",
      "tags": [
        "office"
      ],
      "price": 17.3,
      "quantity": 1
    },
    {
      "name": "binder",
      "tags": [
        "school"
      ],
      "price": 23.36,
      "quantity": 3
    }
  ],
  "couponUsed": false,
  "purchaseMethod": "In store"
}

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

Мы можем начать с $set добавления поля в каждый документ. В сочетании с $sum, мы можем добавить поле с именем itemsTotal в каждый документ.

{ '$set': { 'itemsTotal': { '$sum': '$items.price' } } }

Теперь документы в конвейере были преобразованы и теперь содержат новое свойство с именем itemsTotal.

[
 {
   "_id": "5bd761dcae323e45a93ccffb",
   "items": [
     // ...
   ],
   "itemsTotal": 360.33,
   "couponUsed": false,
   "purchaseMethod": "In store"
 }
]

Далее мы можем передавать документы с $setэтапа на $group этап. Внутри $group мы можем использовать оператор $avg для расчета средней цены транзакции по всем документам.

{ '$group': {
    'averageTransactionPrice': { '$avg': '$itemsTotal' },
    '_id': null
} }

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

[{
 "_id": null,
 "averageTransactionPrice": 620.511328
}]

Выходные данные говорят нам, что средняя цена по всем транзакциям составляет 620,511328 долларов США.

Окончательный код для этого агрегирования в Node.js должен выглядеть примерно так:

const aggCursor = collection.aggregate([
       { '$set': { 'itemsTotal': { '$sum': '$items.price' } } },
       { '$group': { 'averageTransactionPrice': { '$avg': '$itemsTotal' }, '_id': null } }
]);

команда findAndModify

aggregate — не единственная функция, которая пользуется преимуществами синтаксиса агрегирования. Начиная с MongoDB 4.2 , различные команды поддерживают использование конвейеров агрегации для обновления документов.

Давайте взглянем только на одну команду, которая делает это: updateMany.

Возможно, мы захотим добавить itemsTotal постоянное поле в наши документы, чтобы ускорить чтение этих операций.

Давайте воспользуемся updateMany конвейером агрегации, чтобы добавить новое поле с именем itemsTotal.

await collection.updateMany({}, [
   { '$set': { 'itemsTotal': { '$sum': '$items.price' } } },
])

Как вы можете заметить, мы повторно использовали $set сценарий из предыдущего примера. Теперь, если мы проверим нашу коллекцию, мы увидим новое поле в каждом документе.

  {
    "_id": "5bd761dcae323e45a93ccffb",
    "items": [
      {
        "name": "printer paper",
        "price": 17.3,
        // ...
      }
    ],
    "itemsTotal": 360.33,
    "couponUsed": false,
    "purchaseMethod": "In store"
  }

Насколько быстро происходит агрегация MongoDB?

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

Однако это далеко не весь потенциал агрегатного конвейера.

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

Хотя конвейер агрегирования чрезвычайно мощный, насколько он эффективен по сравнению с выполнением такого рода аналитики самостоятельно?

Давайте воспользуемся предыдущим примером запроса агрегации:

const { performance } = require('perf_hooks');
const startTime = performance.now();
const totalAvg = collection.aggregate([
   {
       '$set': {
           'itemsTotal': {
               '$sum': '$items.price'
           }
       }
   }, {
       '$group': {
           '_id': null,
           'total': {
               '$avg': '$itemsTotal'
           }
       }
   }
]);
await totalAvg.toArray()
const endTime = performance.now();
console.log("Aggregation took:", endTime - startTime);

В нашем примере с MongoDB мы используем два этапа: один для добавления itemsTotal поля, а другой для расчета среднего значения itemsTotalпо всем документам.

Чтобы соответствовать этому поведению в Node.js, мы будем использовать Array.prototype.map и Array.prototype.reduce в качестве соответствующих дублеров:

const { performance } = require('perf_hooks');
const startTime = performance.now();
const allItems = await collection.find({}).toArray();
const itemsSum = allItems
   .map(item => {
       item.itemsTotal = item.items.reduce((p, c) => p + parseFloat(c.price), 0);
       return item;
   })
   .reduce((p, item) => {
       return p + item.itemsTotal;
   }, 0);
const itemAvg = itemsSum / allItems.length;

const endTime = performance.now();
console.log("Manual took:", endTime - startTime);

Запуск каждого из приведенных выше фрагментов кода для коллекции из 5000 документов дал следующие результаты по времени:

Агрегация заняла 103,46 мс.

Ручная итерация курсора заняла 881,32 мс.

Это разница более чем в 8,5 раз! Хотя здесь разница может составлять миллисекунды, мы используем чрезвычайно маленький размер коллекции. Нетрудно представить, насколько существенными были бы временные различия, если бы в нашей коллекции содержался миллион или более документов.

Помните, что конвейер агрегации работает на сервере MongoDB и может быть оптимизирован перед запуском, в то время как при переборе курсора для обработки данных на стороне клиента вы добавляете большую задержку из-за выборки страниц данных из этого курсора. Лучший подход, вероятно, представляет собой сочетание того и другого.

Заключение

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