mongodb 중복제거한 아이템 갯수 구하기. (aggregate)

몽고디비에서 종종 중복제거된 아이템 갯수가 필요하다. 

여러개의 도큐먼트가 있고 각각 name이라는 필드값을 갖고있는데, 각각의 필드값은 a일수도, b일수도, c일수도, ...일수도 있다. 여러개의 도큐먼트가 a라는 이름을 갖고 있을수 있고 또 다른 도큐먼트들은 b라는 이름을 갖고있을수 있다. 

 

이 경우 중복되지 않은 name필드의 갯수를 구하고 싶을때 distinct라는걸 쓸수 있다. 

 

https://docs.mongodb.com/manual/reference/method/db.collection.distinct/

{ "_id": 1, "dept": "A", "item": { "sku": "111", "color": "red" }, "sizes": [ "S", "M" ] }

{ "_id": 2, "dept": "A", "item": { "sku": "111", "color": "blue" }, "sizes": [ "M", "L" ] }

{ "_id": 3, "dept": "B", "item": { "sku": "222", "color": "blue" }, "sizes": "S" }

{ "_id": 4, "dept": "A", "item": { "sku": "333", "color": "black" }, "sizes": [ "S" ] }

 

db.inventory.distinct( "dept" )

 

result = [ "A", "B" ]

 

간단하다. 여기에 필터도 걸수 있다. pymongo 예제.

 

# 중복되지 않은 name 필드를 가져오는데 age필드가 40인것만 가져온다.

counts = db[collection].distinct(key='name', filter={'age':40})

 

그리고 이 갯수를 세려면 len(counts)처럼 리스트의 길이를 세주면 된다.(pymongo distinct의 반환값은 리스트.)

 

 

코드를 계속 위와 같이 그냥 distinct에 필터걸어서 사용하고 있었는데, 이게 데이터가 커지니까 메모리가 초과했다고 하면서 실행이 되지 않는다. 그래서 어떻게 해야하나 하고 또 검색을 해봤는데 aggregate프레임워크를 쓰면 된다고 한다. 실제로 아래와 같이 코드를 짰을때 distinct에서는 메모리 에러가 발생하던게 결과가 잘 나옴을 확인할수 있었다.

 

맵리듀스 방식으로 실행시키는 어쩌구 저쩌구... 하는데 맨 아래에 링크 걸어놨는데 그 링크 튜토리얼을 보는게 이해가 가장 쉽다.

 

pipeline = [

    {'$match': {'age':40}},  # 조건

    {'$group': {'_id': '$name', 'count': {'$sum': 1}}}  # 요런식으로 변환해준다.

]

counts = db[collection].aggregate(pipeline)

 

실행시키면 allowDistUse를 True로 설정해주라고 에러가 뜨는 경우가 있는데 

 

counts = db[collection].aggregate(pipeline, allowDiskUse=True)

print len(list(counts))

 

처럼 해주면 된다.

참고로 위에처럼 값을 다 받아와서 list의 길이를 구해도 되지만 aggregate를 썼으니 전부 pipeline으로 처리하자면
https://docs.mongodb.com/manual/reference/operator/aggregation/count/ 처럼 $count를 써서 하자.

 

 

https://docs.mongodb.com/manual/aggregation/

+ 로 중복제거한 아이템을 가져오려면.. 아래처럼 해준다. 위 사진에서 두번째 단계에서 세번째 단계로 갈때 합을 구하는게 아닌 특정 조건으로 중복제거된 후 첫번째 아이템을 가져오는 방법.

https://stackoverflow.com/questions/14184099/fastest-way-to-remove-duplicate-documents-in-mongodb

pipeline = [
                {'$match': {'filter': value}},
                {'$group': {'_id': '$column', "doc": {"$first": "$$ROOT"}}},
            ]
            data = list(self.db[self.collection].aggregate(pipeline, allowDiskUse=True))