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
}

如果你想获得用户，你会：

• 获取u12345的referenceId的所有文档。
• 只获取每种类型的最新版本
• 然后像这样组合并输出用户：

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

我看到的唯一问题是，如果我想按 name 对所有用户进行排序，比如说 - 如果我有 1000 个用户，我看不到这样做的干净方法。

我想知道是否有人对我可以使用的模式有任何建议。我正在使用 MongoDB，所以我可以使用它。

Answer 1

您可以尝试以下聚合。

从类型字段中投影关键字段，按updatedAt和group排序以选择最新值并保留引用和updateat。

对所有文档进行分组并合并不同的键值并保持updatedAt和后处理以格式化文档。

查找以获取用户值，然后使用 replaceRoot 将主文档与查找文档合并。

按名称对文档进行排序。

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

你的问题对具体答案没有足够的要求，所以我会尽量给出一个应该涵盖很多情况的答案。

我怀疑你会找到详细的已发布用例，但是，我可以根据我的个人经验给你一些提示。

高吞吐量：

如果您使用的是高吞吐量事件流，最好将数据存储在事件日志中，其中 ID 不是唯一的并且没有更新，只有插入。这可以通过例如用于事件流的 Kafka 来完成。然后，您可以将事件批量处理到可搜索的数据库中，例如MongoDB。

低吞吐量：

对于较低的吞吐量，您可以将文档直接插入到 MongoDB 中，但是，仍然只插入，而不是更新数据。

在 MongoDB 中以事件日志样式存储数据：

在这两种情况下，在 MongoDB 中，您都需要一个随机的 _id（例如 UUID），因此每个事件都有一个唯一的 _id。要访问逻辑文档，您需要另一个字段，例如docId，与 eventTimestamp 一起将被编入索引（eventTimestamp 排序 desc 以便更快地访问最新版本）。

搜索：

要按其他字段搜索，您可以根据需要使用其他索引，但是，如果您的搜索占用大量 CPU 时间，请确保仅针对 MongoDB 的辅助实例 (secondayOnly) 运行它们，以便事件插入不会耽误的。熟悉 MongoDB 的聚合管道。

为了防止无序更新导致无效状态：

由于您要启用更新，您应该考虑只保存每个文档中的更改，例如+1 字段 A，将字段 B 的值设置为 x。在这种情况下，您需要有一个索引为 docId 和 asc 以 eventTimestamp 结尾，并且不时聚合您的事件到不同集合中的摘要文档中，以便更快地读取最新状态。将docId 的最新文档的eventTimestamp 用于聚合文档，加上aggregationTimestamp 和versionCount。如果您在任何时候收到一个文档，其eventTimestamp 低于聚合集合中最新的eventTimestamp，您将需要部分重新计算该集合。在其他情况下，您可以逐步更新聚合集合。

Answer 3

使用它您将获得所需的输出，确保您已在 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"
            }
        }
    }
])