Сегодня я научился — 02/05/2022
В течение последней недели я использовал MongoDB для сохранения данных, поступающих из темы на Kafka, для каждой темы в Mongo создается новая коллекция. Когда мы хотим получить сообщения темы, нам нужно выполнить постраничный поиск в Mongo, но что если сообщения темы обновляются? Например: платеж, который ожидал подтверждения, а теперь был завершен? У меня будет два сообщения для одной и той же транзакции, я должен взять оба сообщения в постраничный поиск или только при поиске истории платежей мне нужно предыдущее сообщение?
Из сценария, подобного этому, возникла потребность в постраничном поиске в MongoDB, где я мог бы группировать сообщения по последним и выполнять фильтрованный поиск и сортировку.
Для начала нам нужно создать Aggregation, которая представляет собой операцию, обрабатывающую несколько документов и возвращающую результат, она может получать конвейер операций, как это показано в документации mongo в качестве примера:
Внутри нашей агрегации нам понадобится только один $facet в конвейере, поскольку, согласно документации, фасет — это способ обработки нескольких процессов на одном этапе конвейера агрегации, и мы также можем определить, как будет возвращаться результат:
В нашем фасете нам понадобятся два выходных поля (при необходимости их может быть больше, но в моем случае мне понадобилось только два), data, которое содержит документы, и total, которое будет возвращать общее количество документов для заданного фильтра.
Данные:
Данные будут представлять собой конвейер, который сначала будет содержать группу $group, которая работает как порядок по в реляционной базе данных, в этой группе, допустим
{$group: {
_id: '$key',
data: {$last: '$$ROOT'}
}}
поле, которое он будет использовать для объединения документов, является ключом, и он будет вставлять эти сгруппированные документы внутрь данных, беря последний документ группы, который будет самым последним
ps: Порядок операций имеет значение, сначала мы делаем группу, чтобы она могла отделить последнее значение, как последнее значение для входа в mongo
затем мы выполняем сортировку:
{ $sort: { 'data.createdAt': 1 } },
Где мы берем поле createdAt из данных, которые были созданы группой, и говорим, хотим ли мы, чтобы оно было возрастающим или убывающим,
1 для возрастания и -1 для убывания
тогда $skip и $limit
{ $skip: 0},
{ $limit: 10},
Пропустить, чтобы сообщить, сколько документов мы хотим пропустить в пагинации, и ограничение, сколько документов мы хотим привести. Опять же, порядок очень важен, потому что если мы сделаем наоборот в коллекции, которая содержит только 2 документа, поставим limit перед skip и в skip сообщим, что хотим пропустить 1 документ, то произойдет следующее: mongo получит только 1 документ, а skip пропустит его, вернув ноль, в то время как ожидаемое поведение — пропустить первый документ и принести второй.
Граф:
{
$group: {
_id: '$key',
},
В подсчете нам снова нужно сформировать группу, чтобы в подсчете не было больше значений, чем у нас есть на самом деле, а затем мы передаем подсчет, который принимает в качестве аргумента имя, которое мы хотим вернуть в запрос
{ $count: 'total' }
В этом случае он подсчитает, сколько записей было возвращено, и вернет их в качестве общего числа.
В итоге наш код будет выглядеть следующим образом, а если мы хотим сделать какой-то фильтр, то можем использовать оператор $match
$facet: {
data: [
{
$group: {
_id: '$key',
data: {
$last: '$$ROOT',
},
},
},
{ $sort: { 'data.createdAt': 1 } },
{ $skip: 0 },
{ $limit: 10 },
],
count: [
{
$group: {
_id: '$key',
},
},
{ $count: 'total' },
],
},
И возврат, который предоставит mongo, будет следующим, где data содержит самый последний документ
PS: ограничение и пропуск — не лучший способ постраничного поиска. Эта статья посвящена этому вопросу https://scalegrid.io/blog/fast-paging-with-mongodb/.