Spring batch) 대용량 처리를 위한 배치 전략

Spring batch를 사용해서 대용량 데이터에 대한 file to db insert 작업을 수행했다
기존 프로젝트 레퍼런스가 존재하지 않았고 검색 시 spring boot와 jpa 자료는 쉽게 찾아볼 수 있었지만
spirng 자료는 검색 시 활용하기 어려운 자료가 대부분이라 스프링 문서를 읽어가며 로직을 작성했다

---

• 사용 계기

처음엔 엑셀을 읽어 데이터를 list map형태의 파라미터로 만들어 서버로 보내
mybatis의 foreach를 사용해 bulk insert로 처리를 했다
시퀀스 처리와 postgresql에서의 dual 테이블이 제공되지 않아 조금 헤매었지만
몃번 시도 끝에 어렵지 않게 작성할 수 있았디
컬럼이 100개 내외에 row수가 5,000건 이상 되었을 때 isnert 작업이 6분 정도가 소요됐다
절대 많은 데이터가 아니라고 판단되었지만 속도가 왜 이 정도로 느리게 나오는지 파악하기 어려웠다

또한 비교적 긴 시간동안 많은 메모리를 잡아먹어가며 작업이 진행됬기 때문에
작은 오류가 하나 발생하면 어떤 데이터 때문에 오류가 생겼는지 확인이 어려워
작업중인 전체 작업를 중단하고 삭제해 다시 재 작업을 해야 했다

해당 기능은 서비스로 제공되어야 했기 때문에 속도 개선이 무조건 필요했고
검색을 통해 spring batch를 알게 되었다

---

• 문서

공식 페이지의 대량의 레코드를 처리하는 데 필수적인 재사용 가능한 기능을 제공한다고 적혀있다

---

• Chuck Oriented Processing

대용량 작업 시 하나의 트랜젝션 안에서 모든 처리를 할 수 없기 때문에
임의로 chuck단위를 지정해 주고 그 단위 안에서 트랜잭션을 수행한다
때문에 chuck 단위로 커밋이 되고 실패할 경우 해당 chuck만 롤백된다
-> chuck단위를 설정해 주지 않으면 대용량의 데이터중 하나의 작업만 실패해도 전체 데이터를 롤백해야 한다

---

• 수행방법

serviceImpl

    @SuppressWarnings("unchecked")
    @Override
    @Transactional
    public int createJson(Map<String, Object> paramsMap) throws TestException {
    	
    	int isResult = 0;
    	
    	testDao.testSeqInitialization();	//시퀀스 초기화
    	isResult += 1;
    	
    	if(isResult == 1) {
    		if (!((List<HashMap<String, String>>) paramsMap.get("records")).isEmpty()) {
    			testDao.createJson(paramsMap);
    			isResult += 1;
    		}
    	}

    	return isResult;
    }

해당 로직이 트랜잭션 안에서 처리될 수 있게 한다

dao

	/**
	 * 분할 집합
	 * 
	 * @param <T>
	 * @param resList 꼭 분할 집합
	 * @param count   모든 집합 요소 개수
	 * @return 복귀 분할 후 각 집합
	 **/
	public static <T> List<List<T>> split(List<T> resList, int count) {
		
		if (resList == null || count < 1)  return null;
		
		List<List<T>> ret = new ArrayList<List<T>>();
		int size = resList.size();
		if (size <= count) {
			// 데이터 부족 count 지정 크기
			ret.add(resList);
		} else {
			int  pre = size / count;
			int last = size % count;
			// 앞 pre 개 집합, 모든 크기 다 count 가지 요소
			for (int i = 0; i < pre; i++) {
				List<T> itemList = new ArrayList<T>();
				for (int j = 0; j < count; j++) {
					itemList.add(resList.get(i * count + j));
				}
				ret.add(itemList);
			}
			// last 진행이 처리
			if (last > 0) {
				List<T> itemList = new ArrayList<T>();
				for (int i = 0; i < last; i++) {
					itemList.add(resList.get(pre * count + i));
				}
				ret.add(itemList);
			}
		}
		return ret;
	}

chuck 단위를 지정해 주는 로직을 작성해 준다

public int createJson(Map<String, Object> paramsMap) throws TestException {
		
	SqlSession sqlSession = 
    		sqlSessionFactory.openSession(ExecutorType.BATCH);
    	long startTime = System.currentTimeMillis();
    try {
         	List<HashMap<String, String>> list = 
           	(List<HashMap<String, String>>) paramsMap.get("records");
           	List<List<HashMap<String, String>>> ret = split(list, 30);
            for (int i = 0; i < ret.size(); i++) {
                Map<String, Object> params = new HashMap<String, Object>();
                List<HashMap<String, String>> records = ret.get(i);
                params.put("records", records);
                sqlSession.insert("data.test.createJson", params);
        }
        } finally {
            sqlSession.flushStatements();
            sqlSession.close();
        }
        long endTime = System.currentTimeMillis();
        long resutTime = endTime - startTime;
        System.out.println("트랜젝션 배치" + " 소요시간  : " + resutTime / 1000 + "(ms)");
	return 1;

}

배치를 선언해 주고 chuck단위로 돌아갈 수 있도록 로직을 태워준다

---

• 결과

5000 row insert 시 기존 6분에서 1(ms)로 획기적으로 줄어들었다

끝!