몽고DB Aggregate 를 병렬 동작 처럼(Mongodb aggregate facet)

 

MongoTemplate를 활용해서 aggregate를 동작하는 방법은 간단합니다.

아래와 같은 컬렉션을 예를 들어 보겠습니다.

매우 단순한 내용입니다!

 

여기서 우리가 데이터를 집계(aggregate) 하려면 아래와 같은 코드로 동작하게 하였습니다.

public MongoTemplate template;

public void testDb(){
    ProjectionOperation dateProjection = Aggregation.project()
        .and("1").as("1")
        .and("_id").as("_id")
        .and("text").as("text")
        .and("number").as("number")
        .and("type_a").as("type_a")
        .and("type_b").as("type_b");
    Criteria critica = new Criteria().andOperator(
        Criteria.where("type_a").ne(""),
        Criteria.where("type_a").ne(null)
    );
    MatchOperation where = Aggregation.match( critica );
    GroupOperation groupBy = Aggregation.group("1","type_a").sum("number").as("type_a_nums");
    AggregationOptions option = AggregationOptions.builder().allowDiskUse(true).build();
    Aggregation agg = Aggregation.newAggregation(    
        dateProjection,
        Aggregation.sort(Sort.Direction.ASC, "_id"),
        where,	
        groupBy    
    ).withOptions(option);
    
    AggregationResults<HashMap> results = template.aggregate(agg, "test_agg", HashMap.class);
    if(results != null && results.getMappedResults() != null ){
        results.getMappedResults().forEach(arg ->{
            System.out.println(arg);
        });
    }     
}

 

위 코드는 type_a를 group화하여 number의 합계를 구하는 코드입니다.

위 내용을 실행하여 보면,

네...뭐 어렵지 않는 내용입니다.

 

정상적으로 코드가 동작하면 아래처럼 type_a라는 필드를 대상으로 데이터가 묶인결과를 받게 됩니다.

 

그러면 여기서, 데이터가 1억건이 넘는 경우라 가정 하겠습니다.

또한 type_a와 type_b를 각각 group화하여 따로따로 결과를 받아야하는 상황입니다.

다시말해 aggregate에서 group하는 대상이 2가지인 경우 입니다.

이러한 경우에는 통상적으로 aggregate를 두번 실행 하도록 작성하고는 합니다.

public MongoTemplate template;

public void testDb(){

    //첫번째 동작-------------
    ProjectionOperation dateProjection = Aggregation.project()
        .and("1").as("1")
        .and("_id").as("_id")
        .and("text").as("text")
        .and("number").as("number")
        .and("type_a").as("type_a")
        .and("type_b").as("type_b");
    Criteria critica = new Criteria().andOperator(
        Criteria.where("type_a").ne(""),
        Criteria.where("type_a").ne(null)
    );
    MatchOperation where = Aggregation.match( critica );
    GroupOperation groupBy = Aggregation.group("1","type_a").sum("number").as("type_a_nums");
    AggregationOptions option = AggregationOptions.builder().allowDiskUse(true).build();
    Aggregation agg = Aggregation.newAggregation(    
        dateProjection,
        Aggregation.sort(Sort.Direction.ASC, "_id"),
        where,	
        groupBy    
    ).withOptions(option);
    
    AggregationResults<HashMap> results = template.aggregate(agg, "test_agg", HashMap.class);
  
    //두번째 동작-------------
    ProjectionOperation dateProjection222 = Aggregation.project()
        .and("1").as("1")
        .and("_id").as("_id")
        .and("text").as("text")
        .and("number").as("number")
        .and("type_a").as("type_a")
        .and("type_b").as("type_b");
    Criteria critica222 = new Criteria().andOperator(
        Criteria.where("type_b").ne(""),
        Criteria.where("type_b").ne(null)
    );
    MatchOperation where222 = Aggregation.match( critica222 );
    GroupOperation groupBy222 = Aggregation.group("1","type_b").sum("number").as("type_b_nums");
    Aggregation agg222 = Aggregation.newAggregation(    
        dateProjection222,
        Aggregation.sort(Sort.Direction.ASC, "_id"),
        where222,	
        groupBy 222   
    ).withOptions(option);
    
    AggregationResults<HashMap> results222 = template.aggregate(agg222, "test_agg", HashMap.class);    
}

 

이러한 방식으로 첫번째로 type_a에 대해서 aggregate 명령을 수행하게 합니다.

그리고 두번째로 type_b에 대해서 aggregate 명령을 수행하게 합니다.

이렇게 하면 처음 동작 + 두번째 동작에 의해서 시간이 매우 오래 걸리게 됩니다.

 

동작당 3분이라고 가정하면, 6분 정도가 걸리게 될 것 입니다.

만약 데이터가 늘어나고 조회해야 되는 방법이 늘어나게되면 시간은 점점 더 커지게 됩니다.

 

 

* FacetOperation facet!!

몽고DB 3.6 이상부터는 facet 라는 파이프라인 기술이 추가가 되었습니다.

해당 기술은 aggregate의 기능을 대폭 증가시켜주는 기능입니다.

사용법은 아래와 같습니다.

categorize..라는 이름의 형태로 3개의 aggregate 조건을 그룹핑한 모습 입니다.

 

그룹할 동작(Operation)을 작성한 뒤에 key, value형식으로 facet 함수에 넣어준 모습 입니다.

이를 Mongotemplate에 적용하여 보겠습니다.

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

import org.bson.Document;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
import org.springframework.data.mongodb.core.aggregation.AggregationResults;
import org.springframework.data.mongodb.core.aggregation.FacetOperation;

import org.springframework.data.mongodb.core.aggregation.GroupOperation;

import org.springframework.data.mongodb.core.aggregation.MatchOperation;
import org.springframework.data.mongodb.core.aggregation.ProjectionOperation;
import org.springframework.data.mongodb.core.query.Criteria;

public class MongoTemplateFacetTest{

    public MongoTemplate template;

    public void testDb(){

        //첫번째 Operation-------------
        ProjectionOperation dateProjection = Aggregation.project()
            .and("1").as("1")
            .and("_id").as("_id")
            .and("text").as("text")
            .and("number").as("number")
            .and("type_a").as("type_a")
            .and("type_b").as("type_b");
        Criteria critica = new Criteria().andOperator(
            Criteria.where("type_a").ne(""),
            Criteria.where("type_a").ne(null)
        );
        MatchOperation where = Aggregation.match( critica );
        GroupOperation groupBy = Aggregation.group("1","type_a").sum("number").as("type_a_nums");

        //두번째 Operation -------------
        ProjectionOperation dateProjection222 = Aggregation.project()
            .and("1").as("1")
            .and("_id").as("_id")
            .and("text").as("text")
            .and("number").as("number")
            .and("type_a").as("type_a")
            .and("type_b").as("type_b");
        Criteria critica222 = new Criteria().andOperator(
            Criteria.where("type_b").ne(""),
            Criteria.where("type_b").ne(null)
        );
        MatchOperation where222 = Aggregation.match( critica222 );
        GroupOperation groupBy222 = Aggregation.group("1","type_b").sum("number").as("type_b_nums");


        //aggregate 실행 -------------
        AggregationOptions option = AggregationOptions.builder().allowDiskUse(true).build();
        FacetOperation facet = new FacetOperation();  //facet 파이프

        Aggregation agg = Aggregation.newAggregation(
            facet.and(   //and를 통해서 동작할 대상을 넣어줍니다.
                dateProjection,
                Aggregation.sort(Sort.Direction.ASC, "insert_timestamp"),
                where,	
                groupBy
            ).as("first_agg")  //첫번째 행위를 first_agg로 받습니다.
            .and(
                dateProjection222,
                Aggregation.sort(Sort.Direction.ASC, "insert_timestamp"),
                where222,	
                groupBy222
            ).as("second_agg") //두번째 행위를 second_agg로 받습니다.					
        ).withOptions(option);    

        AggregationResults<HashMap> result = template.aggregate(agg, "test_agg", HashMap.class);    
        results.getMappedResults().forEach(arg ->{
            System.out.println(arg.get("first_agg"));
            System.out.println();
            System.out.println(arg.get("second_agg"));
        });   
    }
}    

 

FacetOperation 클래스를 생성한 뒤에  and에 가장 처음 aggregate할 내용을 담아둡니다.

그리고 and를 통해서 두번째로 aggregate를 할 내용을 담아둡니다.

이렇게 and를 통해서 aggregate를 이어 갈 수 있습니다.

물론! 같은 컬렉션에 대해서만 가능합니다.

이를 실행하여보면 정상적으로 나오는 것을 볼 수 있습니다.

결과가 나왔습니다!!!!

 

이렇게 할 경우에 한번에 동작이 이루어지게 되므로 기존에 가정 하였던 6분 걸렸던 내용이 3분으로 줄어들게 됩니다.

여러번의 집계(Aggregate) 동작이 1개의 컬렉션에 의해서 이루어지게 되는 경우에는 이처럼 facet를 통해서 시간을 줄일수가 있습니다!

이상으로 Mongodb에서의 Aggregate의 병렬처리(facet) 방법에 대해서 살펴보았습니다.

틀린부분이나, 궁금한 점은 언제든 연락주세요!

 

* 몽고DB 동시 Aggregate, MongoDB 동시 실행, MongoDB Aggregate 한번에