迭代MongoDB中一个字段中的不同项目答案

【问题标题】：Iterating over distinct items in one field in MongoDB迭代MongoDB中一个字段中的不同项目
【发布时间】：2014-02-22 11:44:34
【问题描述】：

我在 MongoDB 中有一个非常大的集合（约 7M 项），主要由具有三个字段的文档组成。

我希望能够以一种方便的方式迭代其中一个字段的所有唯一值。

目前，我只查询该字段，然后通过迭代游标来处理返回的结果以获得唯一性。这行得通，但它相当慢，我怀疑一定有更好的方法。

我知道 mongo 有 db.collection.distinct() 函数，但这受到最大 BSON 大小 (16 MB) 的限制，我的数据集超过了这个值。

有什么方法可以迭代类似于db.collection.distinct() 的东西，但是使用游标或其他方法，所以记录大小限制不是什么大问题？

我认为也许像 map/reduce 功能这样的东西可能适合这种事情，但我一开始并不真正了解 map-reduce 范式，所以我不知道我在做什么.我正在进行的项目部分是为了学习使用不同的数据库工具，所以我相当缺乏经验。

_{我正在使用 PyMongo，如果它是相关的（我不认为它是）。这应该主要依赖于 MongoDB。}

例子：

对于这个数据集：

{"basePath" : "foo", "internalPath" : "Neque", "itemhash": "49f4c6804be2523e2a5e74b1ffbf7e05"}
{"basePath" : "foo", "internalPath" : "porro", "itemhash": "ffc8fd5ef8a4515a0b743d5f52b444bf"}
{"basePath" : "bar", "internalPath" : "quisquam", "itemhash": "cf34a8047defea9a51b4a75e9c28f9e7"}
{"basePath" : "baz", "internalPath" : "est", "itemhash": "c07bc6f51234205efcdeedb7153fdb04"}
{"basePath" : "foo", "internalPath" : "qui", "itemhash": "5aa8cfe2f0fe08ee8b796e70662bfb42"}

我想做的是迭代只是 basePath 字段。对于上述数据集，这意味着我将迭代 foo、bar 和 baz 各一次。

我不确定它是否相关，但我拥有的数据库是结构化的，因此虽然每个字段都不是唯一的，但所有三个字段的聚合都是唯一的（这是通过索引强制执行的）。

我当前正在使用的查询和过滤操作（注意：我将查询限制为项目的子集以减少处理时间）：

    self.log.info("Running path query")
    itemCursor = self.dbInt.coll.find({"basePath": pathRE}, fields={'_id': False, 'internalPath': False, 'itemhash': False}, exhaust=True)
    self.log.info("Query complete. Processing")
    self.log.info("Query returned %d items", itemCursor.count())
    self.log.info("Filtering returned items to require uniqueness.")
    items = set()
    for item in itemCursor:
        # print item
        items.add(item["basePath"])

    self.log.info("total unique items = %s", len(items))

使用self.dbInt.coll.distinct("basePath") 运行相同的查询会得到OperationFailure: command SON([('distinct', u'deduper_collection'), ('key', 'basePath')]) failed: exception: distinct too big, 16mb cap

好的，这是我最终使用的解决方案。我会将其添加为答案，但我不想减损让我来到这里的实际答案。

    reStr = "^%s" % fqPathBase
    pathRE = re.compile(reStr)
    self.log.info("Running path query")

    pipeline = [
        { "$match" :
            {
                "basePath" : pathRE
            }
        },
        # Group the keys
        {"$group":
            {
                "_id": "$basePath"
            }
        },

        # Output to a collection "tmp_unique_coll"
        {"$out": "tmp_unique_coll"}
        ]

    itemCursor = self.dbInt.coll.aggregate(pipeline, allowDiskUse=True)
    itemCursor = self.dbInt.db.tmp_unique_coll.find(exhaust=True)

    self.log.info("Query complete. Processing")
    self.log.info("Query returned %d items", itemCursor.count())
    self.log.info("Filtering returned items to require uniqueness.")
    items = set()
    retItems = 0
    for item in itemCursor:
        retItems += 1
        items.add(item["_id"])


    self.log.info("Recieved items = %d", retItems)
    self.log.info("total unique items = %s", len(items))

与我以前的解决方案相比，一般性能大约是挂钟时间的 2 倍。在返回 834273 个项目的查询中，具有 11467 个唯一值：

原始方法（检索，填充到 python set 以强制唯一性）：

real    0m22.538s
user    0m17.136s
sys     0m0.324s

聚合管道方法：

real    0m9.881s
user    0m0.548s
sys     0m0.096s

因此，虽然总体执行时间仅缩短了约 2 倍，但聚合管道在实际 CPU 时间方面的性能要高得多。

更新：

我最近重温了这个项目，重写了 DB 层以使用 SQL 数据库，一切都变得简单多了。复杂的处理管道现在是一个简单的SELECT DISTINCT(colName) WHERE xxx 操作。

实际上，MongoDB 和 NoSQL 数据库通常与我在这里尝试做的错误数据库类型相差很大。

【问题讨论】：

可能有一些示例数据？如果我们能看到我们正在尝试做的事情会有所帮助。
@NeilLunn - 这行得通吗？
那种。所以要弄清楚这一点。 “迭代”意味着，您试图将 ("basePath", "internalPath", "itemHash") 的 unique 值组合在一起。甚至限制说foo 炸毁了16MB 的限制。比方说，在aggregate 甚至。这意味着 result 大小。
好吧，我想查询唯一值，然后然后对其进行迭代（有没有一种方法可以在不超过非迭代的 BSON 限制的情况下检索数据？）。是的，即使限制为 basePath 字段也超过了 16 MB 的限制。
澄清我的观点和指定答案的方向，您的 result 集是否可能大于 16MB，您认为 working 设置介于these tolerances 之间或者您已经尝试过吗？

标签： mongodb mongodb-query aggregation-framework pymongo

【解决方案1】：

从目前的讨论点来看，我将对此进行尝试。而且我还注意到，在撰写本文时，MongoDB 的 2.6 版本应该指日可待，如果天气好的话，我将在那里做一些参考。

哦，还有没有在聊天中出现的仅供参考，.distinct() 是一种完全不同的动物，早于此处回复中使用的方法，因此受制于许多限制。

这个解决方案最终是 2.6 或任何当前超过 2.5.3 的开发版本的解决方案

目前的替代方法是使用 mapReduce，因为唯一的限制是输出大小

在不深入了解 distinct 的内部工作原理的情况下，我将继续假设，聚合执行此操作的效率更高 [在即将发布的版本中更是如此]。

db.collection.aggregate([

    // Group the key and increment the count per match
    {$group: { _id: "$basePath", count: {$sum: 1}  }},

    // Hey you can even sort it without breaking things
    {$sort: { count: 1 }},

    // Output to a collection "output"
    {$out: "output"}

])

所以我们使用$out 管道阶段将超过16MB 的最终结果放入它自己的集合中。在那里你可以用它做你想做的事。

由于 2.6 是“指日可待”，因此可以再添加一项调整。

使用runCommand 表单中的allowDiskUse，其中每个阶段都可以使用磁盘并且不受内存限制。

这里的要点是，这几乎可以用于生产。并且性能会比 mapReduce 中的相同操作更好。所以继续玩吧。现在安装 2.5.5 供您自己使用。

【讨论】：

好吧，我想我描述的不正确。我不是要遍历所有将foo 作为basepath 的项目，而是尝试获取一个描述basepath 的所有当前值的集合（没有重复）。基本上，我想要sb.collection.distinct("basePath") 将返回的内容，如果它没有达到 16MB 的限制。如果这是这样做的，我显然真的不了解 mongo。
我可能在这里使用了错误的术语。当我说字段basepath 具有值foo 时，我会说key basepath 具有值foo（例如key : value）。这是否与 MongoDB 描述字段方面的方式不一致（我来自 python dicts 这里）？
@FakeName 那么这会破坏吗？

【解决方案2】：

MapReduce，在当前版本的 Mongo 中可以避免结果超过 16MB 的问题。

map = function() {
    if(this['basePath']) {
        emit(this['basePath'], 1);
    }
    // if basePath always exists you can just call the emit:
    // emit(this.basePath);
};

reduce = function(key, values) {
    return Array.sum(values);
};

对于每个文档，basePath 都会发出一个表示该值计数的值。 reduce 只是创建所有值的总和。生成的集合将具有 basePath 的所有唯一值以及出现的总数。

而且，您需要存储结果以防止使用指定目标集合的out 选项出错。

db.yourCollectionName.mapReduce(
                 map,
                 reduce,
                 { out: "distinctMR" }
               )

【讨论】：

这就是你现在的做法。只是花时间测试正在使用的数据集的限制。
但要注意，这只能通过写入集合来避免问题。所以在调用 mapReduce 命令时使用该选项。

【解决方案3】：

@Neil Lunn 的回答可以简化：

field = 'basePath' # Field I want db.collection.aggregate( [{'$project': {field: 1, '_id': 0}}])

$project 为您过滤字段。特别是，'_id': 0 过滤掉了 _id 字段。

结果仍然太大？用$limit 和$skip 进行批处理：

field = 'basePath' # Field I want db.collection.aggregate( [{'$project': {field: 1, '_id': 0}}, {'$limit': X}, {'$skip': Y}])

【讨论】：

【解决方案4】：

我认为最具扩展性的解决方案是对每个唯一值执行查询。查询必须一个接一个地执行，每个查询都会根据上一个查询结果为您提供“下一个”唯一值。这个想法是查询将返回一个文档，其中将包含您正在寻找的唯一值。如果您使用正确的投影，mongo 将只使用加载到内存中的索引，而无需从磁盘读取。

您可以在 mongo 中使用 $gt 运算符定义此策略，但您必须考虑 null 或空字符串等值，并可能使用 $ne 或 $nin 运算符丢弃它们。您还可以使用多个键来扩展此策略，例如将 $gte 用于一个键，将 $gt 用于另一个键。

此策略应为您提供按字母顺序排列的字符串字段的不同值，或按升序排序的不同数值。

【讨论】：