Разработка управляемой событиями базы данных MongoDB

Question

Цель

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

Я думал о следующей архитектуре:

Справочный документ

_id: "u12345",
type: "user",
createdAt: 1584450565 //UNIX TIMESTAMP

{
  _id: "<random>"
  type: "user-name-revision" //{type}-{key}-Revision
  referenceId: "u12345"
  value: "John Doe Boy"
  updatedAt: 1584450565
}

{
  _id: "<random>"
  type: "user-name-revision"
  referenceId: "u12345"
  value: "John Doe"
  updatedAt: 1584450566 // 1 second higher than the above
}

{
  _id: "<random>"
  type: "user-email-revision"
  referenceId: "u12345"
  value: "john@gmail.com"
  updatedAt: 1584450565
}

Если вы хотите получить пользователя, вы бы:

• Получить все документы с referenceId из u12345 .
• Получите только самые последние данные каждого типа
• Затем объедините и выведите пользователя следующим образом:

_id: "u12345",
type: "user",
createdAt: 1584450565,
name: "John Doe"
email: "john@gmail.com"
updatedAt: 1584450566 // highest timestamp

Единственная проблема, которую я вижу, - это если я хочу отсортировать все пользователи по name скажем - если у меня 1000 пользователей, я не вижу чистого способа сделать это.

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

Answer 1

Вы можете попробовать выполнить агрегацию ниже.

Проецировать ключевое поле из поля типа, отсортировать по updatedAt и group, чтобы выбрать последнее значение и сохранить ссылку и updatedAt. * ​​1003 *

Группировать все документы и объедините различные значения ключа и сохраните updatedAt и постобработку для форматирования документа.

Выполните поиск, чтобы получить пользовательское значение, а затем замените Root, чтобы объединить основной документ с документом поиска.

Сортировка документов по имени.

db.collectionname.aggregate([
  {"$addFields":{"key":{"$arrayElemAt":[{"$split":["$type","-"]},1]}}},
  {"$sort":{"updatedAt":-1}},
  {"$group":{
    "_id":{"referenceId":"$referenceId","key:"$key"},
    "value":{"$first":"$$ROOT"},
    "referenceId":{"$first":"$referenceId"},
    "updatedAt":{"$first":"$updatedAt"}
  }},
  {"$sort":{"updatedAt":-1}},
  {"$group":{
    "_id":"$_id.referenceId",
    "data":{
      "$mergeObjects":{"$arrayToObject":[[["$_id.key","$value"]]]}
    },
    "updatedAt":{"$first":"$updatedAt"}
  }},
  {"$addFields":{
    "data.referenceId":"$referenceId",
    "data.updatedAt":"$updatedAt"
  }},
  {"$project":{"data":1}},
  {"$lookup":{
    "from":"othercollectionname",
    "localField":"data.referenceId",
    "foreignField":"_id",
    "as":"reference"
  }},
  {"$replaceRoot":{
    "newRoot":{
      "$mergeObjects":[{"$arrayElemAt":["$reference",0]},"$data"]}
  }},
  {"$project":{"_id":0}},
  {"$sort":{"name":1}}
])

Альтернативный подход:

При всех преобразованиях ваш запрос будет немного медленнее. Вы можете сделать несколько настроек.

Ввод

{
  _id: "<random>"
  type: "user",
  key: "name"
  referenceId: "u12345"
  value: "John Doe Boy"
  updatedAt: 1584450565
}

Запрос

db.collectionname.aggregate([
  {"$sort":{"updatedAt":-1}},
  {"$group":{
    "_id":{"referenceId":"$referenceId","key":"$key"},
    "top":{"$first":"$$ROOT"}
  }},
  {"$sort":{"top.updatedAt":-1}},
  {"$group":{
    "_id":"$_id.referenceId",
    "max":{"$max":{"$cond":[{"$eq":["$key", "name"]},"$top.value",null]}},
    "key-values":{"$push":{"k":"$_id.key","v":"$top.value"}},
    "updatedAt":{"$first":"$top.updatedAt"}
  }},
  {"$lookup":{
    "from":"othercollectionname",
    "localField":"_id",
    "foreignField":"_id",
    "as":"reference"
  }},
  {"$project":{"_id":0}},
  {"$sort":{"max":1}}
])

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

Ввод

 {
      _id: "<random>"
      type: "user",
      key: "name"
      referenceId: "u12345"
      updates:[
        {"value": "John Doe Boy", updatedAt: 1584450565},
        {"value": "John Doe", updatedAt: 1584450566}
      ]
  }

Запрос

db.collectionname.aggregate([
  {"$addFields":{"latest":{"$arrayElemAt":["$updates",-1]}}},
  {"$group":{
    "_id":"$referenceId",
    "max":{"$max":{"$cond":[{"$eq":["$key", "name"]},"$latest.value",null]}},
    "updatedAt":{"$first":"$updatedAt"}
    "key-values":{"$push":{"k":"$key","v":"$latest.value"}},
    "updatedAt":{"$first":"$latest.updatedAt"}
  }},
  {"$lookup":{
    "from":"othercollectionname",
    "localField":"_id",
    "foreignField":"_id",
    "as":"reference"
  }},
  {"$project":{"_id":0}},
  {"$sort":{"max":1}}
])

Answer 2

Используйте это, вы получите желаемый вывод, убедитесь, что вы проиндексировали в referencedId и updatedAt и достаточно памяти для сортировки.

db.columnName.aggregate([
    {
        $match:{
            referenceId:"u12345"
        }
    },
    {
        $project:{
                type: { $arrayElemAt: [ {$split: [ "$type", "-" ]}, 0 ] },
                referenceId:true,
                createdAt:true,
                name:true,
                email:true,
                updatedAt:true
            }
        },
    },
    {
        $sort:{
            updatedAt:-1
        }
    },
    {
        $group:{
            _id:"$referenceId",
            type:{
                $first:"$type"
            },
            createdAt:{
                $last:"$updatedAt"
            },
            name:{
                $first:"$name"
            },
            email:{
                $first:"$email"
            },
            updatedAt:{
                $first:"$updatedAt"
            }
        }
    }
])

Answer 3

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

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

Высокая пропускная способность:

Если вы используете потоковую передачу событий с высокой пропускной способностью, было бы лучше хранить ваши данные в журнале событий, где идентификаторы не являются уникальными и нет обновлений, только вставки. Это может быть сделано, например, с Kafka, который предназначен для потоковой передачи событий. Затем вы могли бы обработать события в пакетах в доступную для поиска базу данных, например MongoDB.

Низкая пропускная способность:

Для более низкой пропускной способности вы можете вставить документы непосредственно в MongoDB, однако , все еще только вставка, а не обновление данных.

Хранение данных в стиле журнала событий в MongoDB:

В обоих случаях в MongoDB вам понадобится случайный _id (например, UUID), поэтому каждое событие имеет уникальный _id. Для доступа к логическому документу вам понадобится другое поле, например, docId, которое вместе с eventTimestamp будет проиндексировано (с eventTimestamp отсортировано desc для более быстрого доступа к последней версии).

Поиск:

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

Чтобы предотвратить недопустимые состояния из-за неупорядоченных обновлений:

Поскольку вы хотите включить обновления, вам следует рассмотреть только сохраняя изменения в каждом документе, например, +1 в поле A, установите значение x для поля B. В этом случае вам потребуется индекс с docId и как c, заканчивающийся eventTimestamp вместо этого и время от времени объединяйте ваши события в сводные документы в другой коллекции, чтобы обеспечить более быстрое чтение последнего состояния. Используйте eventTimestamp самого последнего документа для docId для агрегированного документа, а также aggregationTimestamp и versionCount. Если в какой-то момент вы получите документ с eventTimestamp ниже, чем последний eventTimestamp в агрегированной коллекции, вам потребуется частично пересчитать эту коллекцию. В других случаях вы можете обновлять агрегированную коллекцию постепенно.