pymongo: usunąć duplikaty (mapa zmniejszyć?)

Question

Mam bazę danych z kilku kolekcji (całkowita ~ dokumentów 15mil) oraz dokumentów wyglądać (uproszczony):

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

Wszyscy mają unikalną pola _id jako dobrze, ale chcę usunąć duplikaty accodring do innego pola (zewnętrznego pola ID).

Po pierwsze, próbowałem bardzo ręcznego podejścia z listami i usuwaniem później, ale DB wydaje się zbyt duży, zajmuje bardzo dużo czasu i nie jest praktyczny.

Po drugie, poniższe wersje nie działają w aktualnych wersjach MongoDB, nawet jeśli ktoś je sugeruje.

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

Więc teraz staram się stworzyć mapę zmniejszyć rozwiązanie, ale ja naprawdę nie wiem, co robię, a zwłaszcza mają trudności z wykorzystaniem innego pola (nie _id baza danych), aby znaleźć i usunąć duplikaty. Oto mój pierwszy złe podejście (przyjęta z jakiegoś źródła interent):

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}}) 

Każda pomoc, aby zmniejszyć wszelkie duplikaty w polu ID zewnętrznej (pozostawiając jednego wpisu), byłyby bardzo apprechiated;) Dzięki!

Answer 1

Alternatywnym podejściem jest użycie modelu o lepszej wydajności niż redukcja map. Rozważmy następujący agregacji rurociąg, który jako pierwszy etap rurociągu agregacji, że $group grupy operatorów dokumentów przez pola ID i przechowuje w dziedzinie unique_ids każda wartość _id zgrupowanych rekordów za pomocą operatora $addToSet. Operator akumulatora sumuje wartości pól przekazanych do niego, w tym przypadku stałej 1 - w ten sposób zliczając liczbę zgrupowanych rekordów w polu liczenia. Drugi etap etapu rurociągu filtruje dokumenty z liczbą co najmniej 2, tj. Duplikatami.

Po uzyskaniu wyniku z agregacją, iteracyjne kursor do usunięcia pierwszego _id w dziedzinie unique_ids, następnie wcisnąć resztę do tablicy, które będą wykorzystywane później do usuwania duplikatów (minus jeden wpis):

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}}) 

Answer 2

Najpierw próbowałem bardzo ręczne podejście z list i usuwanie potem, ale DB wydaje się zbyt duży, trwa bardzo długo i nie jest praktyczne.

Najlepszym sposobem jest użycie metody .aggregate(), która zapewnia dostęp do potoku agregacji w celu znalezienia zduplikowanych dokumentów. Pierwszym etapem w rurociągu jest etapem $group gdzie grupowania dokumentów przez powielane klucz następnie użyj $push i $sum akumulatorowe operatorów, które odpowiednio zwraca tablicę wszystkich _id dla każdej grupy i count elementów w grupie . Następnym i ostatnim etapem procesu jest etap $match, który zwraca tylko te wyniki, w których istnieje duplikat "ID". Następnie należy wykonać iterację kursora i zaktualizować każdy dokument przy użyciu operacji "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 deprecates Bulk() i związane z nim metody więc trzeba będzie użyć metody bulk_write() wykonać żądania.

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) 

---

Można również zrobić to w powłoce, jak pokazano w akceptowanych odpowiedzi na remove dups from mongodb i How to remove duplicates with a certain condition in mongodb?