Java를 활용한 MongoDB 대용량 집계, 억단위

 

 

Mongotemplate를 통해서 데이터를 집계하는 방법은 aggregate라는 함수를 호출하여 실시 한다.

aggregate함수를 적은량의 데이터를 대상으로 실행하면 사실 문제가 되지 않는데, 데이터량이 많아진 경우에 한번에 집계를 실시하면 데이터베이스 서버 또는 동작중인 Java 어플리케이션이 OutofMemory 상태가 될 수 있다.

그럴때를 대비하여 집계함수 실행 간 최소, 최댓값을 설정하여 페이징 처리를 해야된다.

대략적으로 억단위 이상인 데이터에 대해서는 집계함수 실행시 페이징 처리를 해 주어야 문제가 생기지 않는 것 같다.

 

1억건 데이터를 기준으로 1천만건 단위로 집계하여 동작을 시키면 메모리는 2~3기가 정도 사용하며 cpu점유는 10%내외 하드디스크 입출력IO는 40%내외정도로 양호한 편을 보였었다.

was가 아니라 db에서의 자원소모량이며 was에서는 자원소모가 거의 없다.

물론 저정도의 스펙은 일반 컴퓨터(메모리 8G, I5 5세대, HDD)에서 나오는 스펙이므로 좀 더 성능이 좋은 서버에서는 원활할 수도 있다.

 

import java.time.LocalDate;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Optional;

import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.aggregation.AggregationResults;
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.aggregation.SortOperation;
import org.springframework.data.mongodb.core.query.Criteria;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

@Controller
public class MongoTemplateAggTest {
	@Autowired
	private MongoTemplate template;

            @RequestMapping("/testAgg")
            @ResponseBody
	private void aggExample(){	
		ProjectionOperation dateProjection = Aggregation.project()
				.and("date").dateAsFormattedString("%Y-%m-%d").as("daying")
				.and("text1").as("text1")
				.and("text2").as("text2")
				.and("text3").as("text3")
				.and("text4").as("text4")
				.and("_id").as("_id");	
		 
		LocalDate  start_day = LocalDate.parse("2019-06-20");
		LocalDate  end_day = LocalDate.parse("2019-06-28");
		Criteria critica = new Criteria().andOperator(  //날짜로 조회한다는 가정
				Criteria.where("daying").gte(start_day.atTime(0,0,0)).lte(end_day.atTime(0,0,0))  
		);		
		MatchOperation where = Aggregation.match(  //조건절 설정 
				critica
		);
		
		SortOperation sort = Aggregation.sort(Sort.Direction.DESC, "daying");   //억단위가 넘어가는 대용량 데이터에서 정렬하면 에러 또는 일이 안끝날 수도 있다.

		GroupOperation groupBy = Aggregation.group("text1","text2","daying").count().as("count");  //집계할 대상
		
                        final long totalSize = 200000000l;
		final long pagingSize = 10000000l;  //1번 동작할 때 페이지 사이즈
		final int totalCount =(int) (totalSize / pagingSize);          //총 크기에서 한번 페이징할 사이즈를 나눈 값

		List<List<HashMap>> result = new ArrayList<>();
		
		for(long i=0;i < endNumber;i++){
			try {
				Aggregation agg = Aggregation.newAggregation(
						dateProjection,
						where,
						sort,
						Aggregation.limit( pagingSize * (i+1) ),  //집계 페이징
						Aggregation.skip( pagingSize * i  ),						
						groupBy
				);
				AggregationResults<HashMap> res = template.aggregate(agg, "test_collections", HashMap.class);
				if(res != null){
					result.add(res.getMappedResults());
				}				
			} catch (Exception e) {
				e.printStackTrace();
				break;
			}
		}
		Optional.ofNullable(result).ifPresent((res)->{
			res.stream().forEach(System.out::println);
		});
	}

}

먼저 집계할 대상의 조건을 통해서 총 count를 구한다. 그리고 구해진 count를 페이징할 숫자로 나누어 집계함수를 동작시키면 된다.

일반 Mongotemplate에서는 페이징 번호, 총 갯수 형식으로 페이징을 해 주는데 Aggregate에서는 시작값, 끝값 형식으로 구한다.

* 비교

- 일반 페이징 : (1번페이지, 1000개), (2번페이지, 1000개) ...

- 집계 페이징 : (0개, 1000개), (1000개, 2000개) ...