pymongo：删除重复项（地图减少？）

Question

我确实有一个包含多个集合的数据库（总共约 1500 万个文档），文档看起来像这样（简化）：

{'Text': 'blabla', 'ID': 101}
{'Text': 'Whuppppyyy', 'ID': 102}
{'Text': 'Abrakadabraaa', 'ID': 103}
{'Text': 'olalalaal', 'ID': 104}
{'Text': 'test1234545', 'ID': 104}
{'Text': 'whapwhapwhap', 'ID': 104}

它们也都有一个唯一的 _id 字段，但我想根据另一个字段（外部 ID 字段）删除重复项。

首先，我尝试了一种非常手动的方法，使用列表并随后删除，但数据库似乎太大，需要很长时间并且不实用。

其次，以下内容不再适用于当前的 MongoDB 版本，即使有人建议这样做。

db.collection.ensureIndex( { ID: 1 }, { unique: true, dropDups: true } )

所以，现在我正在尝试创建一个 map reduce 解决方案，但我真的不知道我在做什么，尤其是在使用另一个字段（不是数据库 _id）来查找和删除重复项时遇到困难。这是我糟糕的第一种方法（来自一些互联网来源）：

map = Code("function(){ if(this.fieldName){emit(this.fieldName,1);}}")
reduce = Code("function(key,values) {return Array.sum(values);}")
res = coll.map_reduce(map,reduce,"my_results");

response = []
for doc in res.find():
    if(doc['value'] > 1):
        count = int(doc['value']) - 1
        docs = col.find({"fieldName":doc['ID']},{'ID':1}).limit(count)
        for i in docs:
            response.append(i['ID'])

coll.remove({"ID": {"$in": response}})

任何有助于减少外部 ID 字段中的任何重复项（留下一个条目），将非常感激；）谢谢！

Answer 1

我的解决方案也是使用聚合。 您选择为聚合复制的字段。 结果将是重复集合的列表。每个位置将包含一组重复项。 您对列表进行交互，忽略每个组的第一个元素以保留它，并删除其余部分。您为每个重复项集合都这样做。 见下文：

replic = db.<YOUR_COLLECTION>.aggregate([            # Cursor with all duplicated documents
    {'$group': {
        '_id': {'<FIELD_DUPLICATED>': '$<FIELD_DUPLICATED>'},     # Duplicated field
        'idsUnicos': {'$addToSet': '$_id'},
        'total': {'$sum': 1}
        }
    },
    {'$match': { 
        'total': {'$gt': 1}    # Holds how many duplicates for each group, if you need it.
        }
    }
])
                          # Result is a list of lists of ObjectsIds
for i in replic:
    for idx, j in enumerate(i['idsUnicos']):             # It holds the ids of all duplicates 
        if idx != 0:                                     # Jump over first element to keep it
            <YOUR_COLLECTION>.delete_one({'_id': j})     # Remove the rest

您可以尝试“delete_many”来提高性能。

Answer 2

首先，我尝试了一种非常手动的方法，使用列表并随后删除，但数据库似乎太大，需要很长时间并且不实用。

最好的办法是使用.aggregate() 方法，该方法提供对聚合管道的访问，以查找那些重复的文档。管道中的第一阶段是 $group 阶段，您可以在其中按 重复键 对文档进行分组，然后使用 $push 和 $sum 累加器运算符，它们分别返回所有 _id 的数组对于每个组和组中元素的count。管道中的下一个也是最后一个阶段是$match 阶段，仅返回那些存在重复“ID”的结果。然后从那里迭代光标并使用"bulk" 操作更新每个文档。

pipeline = [{'$group': {'_id': '$ID', 'count': {'$sum': 1}, 'ids': {'$push': '$_id'}}},
    {'$match': {'count': {'$gte': 2}}}]

bulk = db.collection.initialize_ordered_bulk_op()
count = 0
for document in db.collection.aggregate(pipeline):
    it = iter(document['ids'])
    next(it)
    for id in it:
        bulk.find({'_id': id}).remove_one({'_id': id})
        count = count + 1
        if count % 1000 == 0:
            bulk.execute()
    if count > 0:
        bulk.execute()

---

MongoDB 3.2 弃用了Bulk() 及其相关方法，因此您需要使用bulk_write() 方法来执行您的请求。

from pymongo import DeleteOne

request = []
for document in db.collection.aggregate(pipeline):
    it = iter(document['ids'])
    next(it)
    for id in it:
        requests.append(DeleteOne({'_id': id}))
db.collection.bulk_write(requests)

---

您也可以在 shell 中执行此操作，如 remove dups from mongodb 和 How to remove duplicates with a certain condition in mongodb? 的已接受答案所示

Answer 3

另一种方法是使用 aggregation framework，它的性能比 map-reduce 更好。考虑以下聚合管道，作为聚合管道的第一阶段，$group 运算符按 ID 字段对文档进行分组，并将每个 _id 值存储在 unique_ids 字段中使用 $addToSet 运算符对记录进行分组。 $sum 累加器运算符将传递给它的字段的值相加，在本例中为常量 1 - 从而将分组记录的数量计算到计数字段中。另一个管道步骤 $match 过滤计数至少为 2 的文档，即重复。

从聚合中获得结果后，您迭代光标以删除 unique_ids 字段中的第一个 _id，然后将其余部分推送到稍后将用于删除重复项的数组中（减去一个条目):

cursor = db.coll.aggregate(
    [
        {"$group": {"_id": "$ID", "unique_ids": {"$addToSet": "$_id"}, "count": {"$sum": 1}}},
        {"$match": {"count": { "$gte": 2 }}}
    ]
)

response = []
for doc in cursor:
    del doc["unique_ids"][0]
    for id in doc["unique_ids"]:
        response.append(id)

coll.remove({"_id": {"$in": response}})