[Spring Batch] Tasklet, Chunk

스프링 배치 동작 구조

step에서 tasklet 방식으로 구성하는 방법과 ItemReader/ItemProcessor/ItemWriter 단위로 Chunk 지향 프로세싱을 구성하는 방법이 있다.

 

Tasklet 방식

 

• 단계 내에서 단일 태스크를 수행하기 위한 것으로 임의의 Step을 실행할 때 읽기/처리/쓰기를 하나의 작업으로 처리하는 방식
• 즉 데이터 처리과정이 tasklet안에서 한번에 이뤄진다
• Job 구성 클래스 외부에 tasklet 클래스를 만들어서 참조하는 방식이다.
  1. Step에서 chunk() 대신 tasklet()을 사용한다.
  2. tasklet은 외부 클래스를 만들고 tasklet 작업을 정의한다.
  3. tasklet 작업을 정의한 외부 클래스에서는 기본적으로 Tasklet을 implements 받아서 구현을 해야한다. 그에 따라 execute 메서드를 구현해야 한다.
     • execute 메서드 인자 contribution/chunkContext
     • contribution - 현재 단계 실행을 업데이트하기 위해 다시 전달되는 변경 가능한 상태
     • chunkContext - 실행 시점의 Job 상태 제공. 호출 간에는 공유되지만 재시작 간에는 공유되지 않는다.
       *tasklet 클래스가 재사용이 될 수 있는 상황에 적합하다.
       
• .배치 처리과정이 쉬운 경우 쉽게 사용되며, 대량처리 경우 더 복잡해질 수 있다.

정리하자면 tasklet은 Job 구성 클래스 내부에 tasklet 구현부를 넣어서 하나의 클래스로 하나의 업무를 정의하는 방식이다.

 

 

 

Chunk 방식

 

• Chunk 기반 : 하나의 큰 덩어리를 n개씩 나눠서 실행
• Chunk 기반 Step은 ItemReader, ItemProcessor, ItemWriter가 있다.
• ItemReader는 배치 처리 대상 객체를 읽어 ItemProcessor 또는 ItemWriter에게 전달한다.
  • 예를 들면, 파일 또는 DB에서 데이터를 읽는다.
• ItemProcessor는 input 객체를 output 객체로 filtering 또는 processing 해 ItemWriter에게 전달한다.
  • 예를 들면, ItemReader에서 읽은 데이터를 수정 또는 ItemWriter 대상인지 filtering 한다.
  • ItemProcessor는 optional 하다.
  • ItemProcessor가 하는 일을 ItemReader 또는 ItemWriter가 대신할 수 있다.
• ItemWriter는 배치 처리 대상 객체를 처리한다.
  • 예를 들면, DB update를 하거나, 처리 대상 사용자에게 알림을 보낸다.

 

 

Chunk 방식으로 데이터처리

// step를 chunk로 생성해보기
@Bean
public Step chunkBaseStep(){
    return stepBuilderFactory.get("chunkBaseStep")
            .<String, String> chunk(10)
            .reader(itemReader())
            .processor(itemProcessor())
            .writer(itemWriter())
            .build();
}

private ItemWriter<String> itemWriter() {
    return items -> log.info("chunk item size: {}", items.size());
}

private ItemProcessor<String, String> itemProcessor() {
    //return 값이 null이면 wrtier로 넘어가지않는다.
    return item -> item + "spring batch";
}

private ItemReader<String> itemReader() {
    //ListItemReader는 스프링에서 기본적으로 제공.  리스트를 받아서100개의 read한다.
    return new ListItemReader<>(getItems());
}

chunksize로 처리 시, Step에 .chunksize(크기) 로 정의하며 앞에 처리될 타입을 정의한다.

Processor에서는 앞서 말한것과 같이 <reader에서 읽은 타입, writer에 보낼 타입> 으로 정의해 처리한다.
<String,String>
앞의 String은 reader에서 읽은 데이터의 타입이고
뒤의 String은 writer에서 받을 데이터의 타입이다.
따라서 processor 처리 함수에서는 앞/뒤 타입을 모두 선언해서 필터나 데이터 변형 처리가 가능하다.

Reader에서는 리턴으로 선언한 ListItemReader는 spring batch에서 제공하는 item처리 클래스이다.
getItems에서 가져온 List<String>을 LIstItemReader에서 알아서 item반환해준다.

Writer에서는  <String>이지만, List<String>단위로 처리되며,
items를 받아 log를 출력하면 chunksize=10개씩 읽은 String이 List로 출력된다.

 

 

Tasklet 방식으로 데이터처리

 

@Bean
public Step taskBaseStep(){
    return stepBuilderFactory.get("taskBaseStep")
            .tasklet(this.tasklet())
            .build();
}

// chunk 방식을 tasklet 방식으로 구현
private Tasklet tasklet() { //Tasklet 클래스 설정, TaskletStepBuidler를 반환한다.
    List<String> items = getItems();
    return (contribution, chunkContext) -> {
        StepExecution stepExecution = contribution.getStepExecution();

        int chunkSize = 10;
        int fromIndex = stepExecution.getReadCount();
        int toIndex = fromIndex + chunkSize;
        if (fromIndex >= items.size()){
            return RepeatStatus.FINISHED;
        }
        List<String> subList = items.subList(fromIndex, toIndex);

        log.info("chunk item size: {}", subList.size());

        stepExecution.setReadCount(toIndex);

        return RepeatStatus.CONTINUABLE;
        //만약 fromIndex 가 10이고, toIndex가 10이면 인덱스10번부터 10개의 item를 꺼낼수 있다는뜻

    };
}

 

tasklet의 contributution에서 StepExecution 으로 step이 얼마나 처리되고 있는지의 정보를 알 수 있다.

(chunksize를 10으로 정의하고 readCount로 페이징을 처리)

 

비교적 대량 데이터는 chunk방식으로 처리하는게 더 간단하다.

 

 

참조

https://choisblog.tistory.com/80
https://velog.io/@gkskaks1004/%EC%8A%A4%ED%94%84%EB%A7%81-%EB%B0%B0%EC%B9%98%EC%9D%98-Tasklet-%EB%B0%A9%EC%8B%9D%EA%B3%BC-Chunk-Tasklet-%EB%B0%A9%EC%8B%9D