Chunk 프로세싱

SpringBatch의 chunk 프로세싱

100개의 row 데이터는 한번에 처리하는 것이 충분히 가능하지만, 100만개의 row 데이터는 조금 어려움이 있을 것이다.

• 메모리상에 올려야 하는 문제

• 100만개의 데이터를 실행하기 위한 connection 문제 등

큰 데이터를 Chunk 단위로 분할 처리하는 것이 Chunk 프로세싱이다.

Chunk 프로세싱 배치 플로우

1. Reader로 응답 데이터를 읽어와서 read() 함수를 통해 한 row씩 응답.

2. 응답 받은 데이터를 Processor에서 가공 처리 한다.

3. 가공된 데이터를 ChunkSize만큼 모아, Writer에서 저장소에 저장한다.

Chunk 프로세싱 의사 코드

void Chunk_처리_방법(int chunkSize, int totalSize) {
   for (int i = 0; i < totalSize; i = i + chunkSize()) {
	List items = new ArrayList<>();
	for (int j = 0; j < chunkSize; j++) {
	   Object item = ItemReader.read();
	   Object processedItem = itemProcessor.process(item);
   	   items.add(processedItem);
	}
	itemWriter.write(items);
   }
}

1. 첫 for문 안에서 Writer에서 처리할 Item list를 생성한다.

2. 두 번째 for문에서는 chunkSize만큼 loop를 돌면서, ItemReader에서 Item을 가져온다.

   • Processor가 있다면 가공 처리를 진행함. (Processor가 필수 요소는 아니다.)

   • 가공 처리가 끝나면 아이템에 추가한다.

3. ItemWriter에서 Item을 저장 처리한다.

위 과정이 ChunkSize의 한 사이클이 수행되는 것이다. 이 작업을 totalSize만큼 작업한다.

쉽게 요약하면 Reader에서 데이터를 하나 읽고, 그 데이터를 Processor를 통해 처리한다.

처리된 데이터들이 Chunk 단위만큼 쌓이면 Writer에서 저장소에 write( )한다.

• Reader, Procssor에서는 하나씩, Writer에서 Chunk 단위 처리

여기서 중요한 것은 데이터를 읽어서 Chunk라는 단위로 트랜잭션을 수행한다는 것이다.

Job은 어떻게 실행되는가 - 트랜잭션 관련

https://devocean.sk.com/blog/techBoardDetail.do?ID=164085

public Step step(.., PlatformTransactionManager transactionManager) {
    ...
        .<>chunk(1000, transactionManager)
        .reader()
        .processor()
        .writer()
        .build();
}

read()
트랜잭션 매니저.doTransaction start
process()
writer()
트랜잭션 매니저.commit()

Job을 실행할 때 JobLauncher를 통해 실행하는데, 메소드를 따라 들어가면 TaskletStep 클래스의 doExecute()가 존재한다. 해당 메소드는 Job안에 Step이 정의되어 있는데 그 Step이 호출할 때 실행된다.

@Override
protected void doExecute(StepExecution stepExecution) throws Exception {
    ...
    stepOperations.iterate(new StepContextRepeatCallback(stepExecution) {

	@Override
	public RepeatStatus doInChunkContext(RepeatContext repeatContext, ChunkContext chunkContext)
			throws Exception {

	   StepExecution stepExecution = chunkContext.getStepContext().getStepExecution();

	   // Before starting a new transaction, check for
	   // interruption.
   	   interruptionPolicy.checkInterrupted(stepExecution);

	   RepeatStatus result;
	   try {
	       result = new TransactionTemplate(transactionManager, transactionAttribute)
		    .execute(new ChunkTransactionCallback(chunkContext, semaphore));
	   }
	   catch (UncheckedTransactionException e) {
		// Allow checked exceptions to be thrown inside callback
		throw (Exception) e.getCause();
	}
}

Step의 ChunckContext를 Srping Trasaction의 TransactionTemplate 위에서 Callback 형태로 동작한다.

결국 Step의 Chunk Tasklet은 transactionManager 위에서 실행되고 있다.

Job이 실행되고 read 하기 전부터 transaction이 시작되고, 종료 전까지 장애가 발생하면 모두 rollback 시켜버린다.