MongoDB aggregate (feat: spring batch partition)

 

서론

---

 

MongoDB에서는 find에서 사용되는 일반적인 문법으로는 RDBMS에서 지원하는 group by, join 과 같은 데이터 조회문을 만들 수 없다, 대신 MongoDB에서는 aggregate 라는 기능을 통해 보다 복잡한 데이터 조회문을 작성할 수 있게 해준다.

 

 

지금까지는 aggregate 기능을 사용할 일이 없었는데 이번 배치 작업에서 사용할 일이 생겨서 이번 기회에 aggregate에는 어떤 기능들이 있고 나는 어떻게 사용했는지 간단히 정리할 예정이다.

 

 

 

본론

---

 

 

이번 글의 목차는

• aggregate 알아보기
• 사용 예시 (with Mongo Template)

으로 구성될 예정이다.

 

 

aggregate 알아보기

aggregate는 위에서도 말한 것처럼 보다 복잡한 질의문을 지원해주고 파이프라인을 만들어서 각각 연산 단계를 구분할 수 있다.

db.sales.aggregate([
    {
        // 1. 특정 기간 동안의 판매만 필터링
        $match: {
            date: { 
                $gte: ISODate("2023-01-01"), 
                $lt: ISODate("2023-12-31") 
            }
        }
    },
    {
        // 2. 고객별로 총 판매 금액을 계산
        $group: {
            _id: "$customer_id",
            totalSales: {
                $sum: { $multiply: ["$price", "$quantity"] }
            }
        }
    },
    {
        // 3. 총 판매 금액을 기준으로 내림차순 정렬
        $sort: { totalSales: -1 }
    },
    {
        // 4. 상위 5명의 고객만 선택
        $limit: 5
    }
])

요런 느낌이다.

 

 

MongoDB는 mapReduce라는 기능도 제공하는데 공식문서 를 확인해보면 mongo 5.0부터는 더이상 사용하지 않는다고 한다, mapReduce는 js 기반으로 돌아가다보니 성능쪽으로 이슈가 많아 c++로 개발되어 mongoDB에 내장된 aggregate가 대체제로 나왔기 때문이다.

 

 

(aggregate로 안되는 경우 mapReduce를 사용해야겠지만 그럴 때는 그 많고 복잡한 내용을 쿼리 하나에 몰아넣는게 맞을까 돌아보는게좋지 않을까..?)

 

 

 

알아두면 좋을만한 대표적인 aggregate에서 지원하는 문법들을 보면

연산자	설명	예시
$match	쿼리 조건에 맞는 문서 필터링. find와 유사하게 동작.	{ $match: { status: "active" } }
$group	필드 값을 기준으로 그룹화하고, 그룹별로 집계 연산을 수행.	{ $group: { _id: "$customerId", total: { $sum: "$price" } } }
$project	출력할 필드 지정. 새로운 필드를 추가하거나, 기존 필드를 변환 가능.	{ $project: { name: 1, totalPrice: { $multiply: ["$price", "$quantity"] } } }
$sort	필드 값을 기준으로 정렬.	{ $sort: { totalSales: -1 } }
$limit	결과에서 제한된 수의 문서만 반환.	{ $limit: 10 }
$skip	지정된 수의 문서를 건너뜀.	{ $skip: 5 }
$unwind	배열 필드를 개별 문서로 분리. 배열의 각 요소마다 하나의 문서를 생성.	{ $unwind: "$items" }
$lookup	다른 컬렉션과 조인(Join)을 수행.	{ $lookup: { from: "orders", localField: "custId", foreignField: "orderId", as: "orderData" } }
$bucket	값을 기준으로 데이터를 구간별로 그룹화.	{ $bucket: { groupBy: "$price", boundaries: [0, 100, 200, 300], default: "Other", output: { count: { $sum: 1 } } } }
$bucketAuto	자동으로 구간을 나누어 데이터를 그룹화.	{ $bucketAuto: { groupBy: "$age", buckets: 5 } }

 

정도가 있다.

 

 

사용 예시 (with Mongo Template)

 

 

나는 특정 컬렉션의 모든 데큐먼트를 조회하고 필드 값에 따라 특정 로직을 수행하는 기능을 작성해야 했다, 데이터가 적을 경우에는 문제가 없지만 데이터가 많아질 경우 처음 읽은 도큐먼트와 나중에 읽은 도큐먼트에 대해 로직 수행에 있어서 시간 차이가 날거 같아 파티셔닝을 수행하였다.

 

 

파티셔닝을 할 때 aggregate중 bucketAuto를 사용하였는데, 해당 컬렉션을 일정한 사이즈로 나눠 거기에 해당하는 모든 도큐먼트를 조회해야 했기에 pk 인덱스를 사용하고자 id를 기반으로 나눴다.

BucketAutoOperation auto = Aggregation.bucketAuto("_id", gridSize)
            .andOutput("_id").max().as("end_id")
            .andOutput("_id").min().as("start_id");

Aggregation aggregation = Aggregation.newAggregation(auto);

AggregationResults<PartitionResult> results = mongoTemplate.aggregate(
    aggregation, "colleciton", ResultDTO.class);

List<ResultDTO> mappedResults = results.getMappedResults();

 

위 코드는 내가 사용했던 aggregate를 비슷하게 재현한 코드이다.

 

 

aggregate(aggregate 문, 컬렉션 이름, 결과 맵핑할 클래스) 와 같은 방식으로 수행할 수 있고 이렇게 하면
AggregationResults 라는 클래스를 반환하는데 내부에는

• mappedResults -> 맵핑 클래스에 값 맵핑한 결과 (List, 단건 클래스 둘 다 가능)
• rawResults -> mongoDB 에서 반환한 결과 그대로 (디버깅용?)

정도가 있다.

 

 

 

이때 주의할 점이 있는데 보통 mongoDB를 사용할 때 naming strategy config를 설정할 것이다. 우리 팀의 경우 db convention이snake case라 설정을 해주는데 저 as("end_id") 에는 적용이 안됐다. 그래서 확인해보니 aggregate 코드의 as에서는 해당 변환이 적용이 안된다고 하니 잘 맞춰서 사용하자.

 

 

(DTO에 @Field(name = "endId") 이런식으로 어노테이션을 사용하여 맵핑하는 것도 방법이다.)

 

 

 

이렇게 ResultDto 리스트를 받아온 후 Execution Context에 startId, endId를 넣어서 파티셔닝 기반으로 조회가 잘 되는 것을 확인하였다.

 

 

결론

---

글 쓸 소재가 없어서 어떻게든 쓰다 보니 방향성이 좀 애매하긴 한데 누군가에게는 도움이 됐으면 좋겠다..!