Базовая агрегация в MongoDB 2.1 с Python

К сожалению, из-за того, что MongoDB использует движок Javascript SpiderMonkey , каждый процесс mongod ограничен одновременным выполнением одного потока Javascript. Таким образом, для того чтобы получить какой-либо параллелизм с помощью MongoDB mapreduce (), вы должны запустить его на кластере с сегрегацией, а на кластере с N сегментов — вы ограничены N-сторонним параллелизмом.

Новая структура агрегации, с другой стороны, реализована на C ++ и имеет собственный язык на основе BSON, поэтому вы вообще не обращаетесь к интерпретатору Javascript.

Mapreduce на самом деле не может сделать все

Если вы читали мою предыдущую статью об агрегации MongoDB , вы могли заметить, что формат вывода команды mapreduce () довольно ограничен и всегда имеет форму

{ _id: ..., /* some (possibly compound) key */
  value: ... /* some (possibly compound) value */
  }

Но что, если вам нужны результаты в каком-то другом формате? С mapreduce вы застряли.

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

Новая структура агрегации, с другой стороны, имеет поэтапные вычисления в основе, а конвейер — фундаментальную структуру агрегации.

Серия труб

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

Еще один способ взглянуть на новую структуру агрегации — это «супер-находка». Агрегатная команда может сделать что-то другое, чем метод find (), и даже больше. Например, следующий запрос:

q = db.my_collection.find({...some query...}, {...some fields...})
q = q.sort(...some sort...)
q = q.skip(...some skip...)
q = q.limit(...some limit...)

будет реализован следующим совокупным конвейером:

pipeline = [
    { '$match': ...some query... },
    { '$sort': ...some sort...},
    { '$skip': ...some skip... },
    { '$limit': ...some limit... }
    { '$project': ...some fields...},
    ]
q = db.command('aggregate', 'my_collection', pipeline=pipeline )

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

Выше, конвейер (концептуально) делает следующее:

• Для каждого документа в my_collection передайте только те документы, которые $ соответствуют данному запросу
• $ сортировать все документы, выданные на данный момент, с помощью ключа сортировки, а затем отправлять отсортированные документы вместе
• $ пропускает указанное количество документов, не отправляя их на последующие этапы конвейера, а затем отправляет остальные по конвейеру.
• Отправляйте только указанное количество документов в $ limit на следующие этапы конвейера.
• $ проецирует документы в новый формат, отправляя по каждому повторно сформированному документу

В конце у нас есть что-то, что дублирует вывод запроса выше. Теперь, если бы мы решили использовать $ sort, а затем $ match, сервер мог бы иметь еще несколько документов для сортировки в $.

На самом деле агрегирование

Хотя хорошо, что MongoDB переросла операция «супер-поиск», наши конвейеры агрегации могут сделать гораздо больше. Предположим, что мы хотим вычислить среднее количество значений x, сгруппированных по значениям y. Сначала мы вставим некоторые случайные данные для тестирования (я использую мой сервер MongoDB на порту 27018, а не по умолчанию 27017):

import pymongo
import random
conn = pymongo.Connection(port=27018)
db = conn.agg
db.data.insert([
...     dict(x=random.random(), y=random.randint(0,10))
...     for i in range(50)])
[ObjectId(...)...]

Базовая группировка и формирование документов

Теперь, чтобы вычислить среднее значение, мы можем использовать оператор конвейера $ group:

pipeline = [
...     {'$group': {
...         '_id': '$y',
...         'mean': {'$avg': '$x'} } } ]
db.command('aggregate', 'data', pipeline=pipeline)
{u'ok': 1.0, u'result': [
    {u'_id': 8, u'mean': 0.45158769756086975}, 
    {u'_id': 7, u'mean': 0.10767920380630946}, 
    {u'_id': 9, u'mean': 0.537524071579183}, 
    {u'_id': 5, u'mean': 0.4712010849527597}, 
    {u'_id': 10, u'mean': 0.6450144108731366}, 
    {u'_id': 3, u'mean': 0.5829812441520267}, 
    {u'_id': 2, u'mean': 0.5139874597597852}, 
    {u'_id': 1, u'mean': 0.35474322219101523}, 
    {u'_id': 4, u'mean': 0.6813855492820068}, 
    {u'_id': 0, u'mean': 0.47123430216880813}]}

Первое, что нужно отметить выше, это использование $ x и $ y. Это синтаксис структуры агрегации для ссылки на поля в документе. Это необходимо, потому что мы можем также выполнять некоторые простые операции с документами. Например, предположим, что мы хотели удвоить значения x. Для этого мы бы использовали оператор $ project:

pipeline = [
...     { '$project': {
...         '_id': 0,
...         'x': p1,
...         'y': 1,
...         'double_x': {
...             '$multiply': [ '$x', 2 ] 
...         } } } ]
db.command('aggregate', 'data', pipeline=pipeline)
{u'ok': 1.0, u'result': [
    {u'y': 8, u'x': 0.0731577856277077, 
     u'double_x': 0.1463155712554154}, 
    {u'y': 8, u'x': 0.6628554147686313, 
     u'double_x': 1.3257108295372626}, 
    ... ] }

Разматывать массивы

Еще одна приятная вещь, которую мы можем сделать, — это рассматривать элементы встроенного массива в документе как сами документы. Предположим, у нас есть несколько комментариев в блоге, например:

db.blog.articles.insert( [ 
...     { 'title': 'First Post',
...       'comments': [
...            {'author': 'Someone'},
...            {'author': 'Someone Else'},
...            {'author': 'Someone'},
...            {'author': 'Someone'} ] },
...     { 'title': 'Second Post',
...       'comments': [
...            {'author': 'Another commenter'},
...            {'author': 'Someone Else'},
...            {'author': 'Someone'} ] } ] )
[ObjectId(...), ObjectId(...)]

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

pipeline = [ {'$unwind': '$comments'} ]
db.command('aggregate', 'blog.articles', pipeline=pipeline)
{u'ok': 1.0, u'result': [
    {u'_id': ObjectId(...), 
     u'comments': {u'author': u'Someone'}, 
     u'title': u'First Post'},
    {u'_id': ObjectId('...'), 
     u'comments': {u'author': u'Someone Else'}, 
     u'title': 
     u'First Post'}, ... ] }

Обратите внимание, что документы, выходящие из $ unwind, имеют ту же структуру, что и входные данные, за исключением массива unwound, который был заменен одним элементом. После того, как у нас есть свернутые документы, мы можем добавить некоторые другие операторы для вычисления количества постов на комментатора:

pipeline = [ 
...     {'$unwind': '$comments'},
...     {'$group': {
...         '_id': '$comments.author',
...         'comments': { '$sum': 1 } } } ]
db.command('aggregate', 'blog.articles', pipeline=pipeline)
{u'ok': 1.0, u'result': [
    {u'_id': u'Another commenter', u'comments': 1}, 
    {u'_id': u'Someone Else', u'comments': 2}, 
    {u'_id': u'Someone', u'comments': 4}]}

Стоит отметить, что мы можем использовать $ group в качестве некоего обратного отката, используя функцию аггрегирования $ push. Возвращаясь к нашему синтетическому (x, y) примеру, мы можем «свернуть» все значения x для данного y:

pipeline = [
...     {'$group': {
...         '_id': '$y',
...         'xs': {'$push': '$x'} } },
...     {'$project': {
...         '_id': 0,
...         'y': '$_id',
...         'xs': 1 } },
...     {'$sort': 'y'} ]
db.command('aggregate', 'data', pipeline=pipeline)
{u'ok': 1.0, u'result': [
    {u'y': 0, u'xs': [
        0.5480646819579362, 0.06989250292970206, 
        0.1483275969049508, 0.37886136252339875, 
        0.944175551482085, 0.6528854869305418, 
        0.6699875222192363, 0.038172503680066416, 
        0.29335286756814327, 0.8896985443297418, 
        0.5501587033310872] }, 
    ... ] }

Здесь мы сначала $ grouped для вычисления интересующего нас агрегата, затем применили $ project, чтобы немного переименовать поля, заканчивая $ sort, чтобы представить результаты в отсортированном порядке.

Новая агрегация не может делать все (пока)

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

Нет оператора $ out

К сожалению, запланированный оператор $ out не попал в версию агрегации 2.1. Это означает, что вы не можете определить конвейер и поместить его результаты в другую коллекцию; Вы должны быть готовы обработать результаты, когда команда вернется. Из-за этого результаты агрегирования ограничены 16 МБ из-за ограничения размера документа в MongoDB. Так что, если вам нужно вычислить миллионы групп, у вас будет один из следующих вариантов:

• Разбейте ваш конвейер с помощью $ match или $ skip / $ limit, чтобы результаты каждого конвейера были ниже 16MB
• Откат к карте
• Вывод данных из MongoDB и агрегирование в какой-то внешней системе (такой подход используется , например, в Zarkov ). все вычисления происходят на сервере.

Там нет объяснения ()

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

• Поставьте ваш оператор $ match первым . Этот оператор может использовать индекс (yay!), И вы можете проверить это, используя тот же запрос с find (). Объяснение ().
• Если возможно, поместите ваш $ sort сразу после $ match. Это приводит к тому, что первые два этапа ведут себя так же, как find (). Sort (), что опять-таки можно объяснить ().
• Будьте осторожны с операторами с сохранением состояния. Большинство операторов конвейера не имеют состояния , поэтому их использование памяти не меняется в зависимости от длины конвейера. Но $ group и $ sort — нет, поскольку они накапливают результаты в оперативной памяти. Так что будьте осторожны, где вы их положили.

В новой структуре агрегации есть ряд других функций, которые я здесь не рассматривал. Для получения более подробной информации посетите официальные документы . Мне было бы интересно услышать, что вы думаете о новой структуре агрегации, особенно если кто-то использует ее в процессе производства (или подготовки к работе). Позвольте мне знать в комментариях ниже!