섹션 7. 스프링 배치 청크 프로세스 이해

[0JUUU]

• [x] Chunk
• [x] ChunkOrientedTasklet - 개념 및 API 소개
• [x] ChunkOrientedTasklet - ChunkProvider / ChunkProcessor
• [x] ItemReader / ItemWriter / ItemProcessor 이해
• [x] ItemStream
• [x] Chunk Process 아키텍처

[0JUUU]

Chunk

• 여러 개의 아이템을 묶은 하나의 덩어리, 블록
• 한 번에 하나씩 아이템을 입력받아 Chunk 단위의 덩어리로 만든 후 Chunk 단위로 트랜잭션을 처리 (Chunk 단위의 Commit / Rollback이 이루어짐)
• 대용량 데이터를 한 번에 처리하는 것이 아닌 Chunk 단위로 쪼개 더 이상 처리할 데이터가 없을 때까지 반복해서 사용

Chunk<I> vs Chunk<O>

• Chunk<I> : ItemReader로 읽은 하나의 아이템을 Chunk에서 정한 개수만큼 반복해서 저장
• Chunk<O> : ItemProcessor에서 적절하게 가공, 필터링 ➡️ ItemWriter에 전달

1. ItemReader가 Source를 하나씩 읽고 하나씩 Chunk<I>에 저장 ➡️ 이를 Chunk 크기만큼 실행
2. Chunk 크기만큼 쌓였다면 Chunk<I>를 ItemProcessor에 전달
3. ItemProcessor: 전달받은 Chunk(list) 를 적절하게 가공해서 Chunk<O>에 저장
   • 이 때 아이템들을 하나씩 읽어 처리한다.
4. Chunk<O>를 ItemWriter에 전달
5. ItemWriter: 데이터 쓰기 작업

[0JUUU]

ChunkOrientedTasklet - 개념 및 API 소개

• 스프링 배치에서 제공하는 Tasklet의 구현체 ➡️ Chunk 지향 프로세싱 담당
• ItemReader, ItemWriter, ItemProcessor를 사용하여 Chunk 기반의 데이터 입출력 처리 담당
• TaskletStep에 의해 반복적으로 실행, 실행될 때마다 매번 새로운 트랜잭션 생성
  • exception 발생 ➡️ 해당 Chunk 롤백 & 이전 커밋한 Chunk 완료된 상태로 유지
• 내부적으로 ItemReader를 핸들링하는 ChunkProvider와 ItemProcessor, ItemWriter를 핸들링하는 ChunkProcessor 타입의 구현체를 가짐

1. TaskletStep: execute 메서드 ➡️ ChunkOrientedTasklet 호출
2. ChunkOrientedTasklet: provide 메서드 ➡️ ChunkProvider 호출
3. ChunkProvider: ItemReader에게 Item을 하나씩 read하도록 지시
   • 이 것을 Chunk Size만큼 반복
4. ChunkOrientedTasklet: ChunkProcessor에게 읽은 데이터를 가공하라고 명령 (앞서 read한 내용들이 inputs로 들어감)
5. ChunkProcessor: ItemProcessor에게 가공하라고 명령
   • 이것을 전달된 아이템 개수만큼 반복
6. ChunkProcessor: 가공된 아이템을 ItemWriter에게 전달
7. ItemWriter: 저장하는 등의 쓰기 처리
8. 위의 일련의 과정이 하나의 사이클로 이후 ChunkOrientedTasklet에 가서 읽을 Item이 없을 때까지 반복

// step1 구성
    @Bean
    @JobScope
    public Step step1() {
        return stepBuilderFactory.get("step1")
                                 .<String, String>chunk(2)
                                 .reader(new ListItemReader<>(Arrays.asList("item1", "item2", "item3", "item4", "item5", "item6")))
                                 .processor(new ItemProcessor<String, String>() {
                                     @Override
                                     public String process(String item) throws Exception {
                                         return "my_" + item;
                                     }
                                 })
                                 .writer(new ItemWriter<String>() {
                                     @Override
                                     public void write(List<? extends String> items) throws Exception {
                                         items.forEach(System.out::println);
                                         System.out.println("=============");
                                     }
                                 })
                                 .build();
    }

ChunkSize : 3

ChunkSize : 2

[0JUUU]

ChunkOrientedTasklet - ChunkProvider / ChunkProcessor

ChunkProvider

• ItemReader를 사용 ➡️ 소스로부터 아이템을 읽어 Chunk 단위로 만들어 제공하는 도메인 객체
• Chunk<I> 만들고 내부적으로 반복문을 사용하여 ItemReader.read()를 계속 호출 ➡️ item을 Chunk에 쌓음
  • Chunk Size 만큼 Item을 읽으면 반복문 종료 & ChunkProcessor로 넘김
  • ItemReader가 읽은 item이 null인 경우, read 반복문 종료 & 해당 Step의 반복문도 종료
• 외부로부터 ChunkProvider가 호출될 때마다 새로운 Chunk 생성

ChunkProcessor

• ItemProcessor를 사용해서 Item을 변형, 가공, 필터링 / ItemWriter를 사용해서 Chunk 데이터를 저장, 출력
• Chunk<O>를 만들고 앞에서 넘어온 Chunk<I>의 item을 한 건씩 처리한 후 Chunk<O>에 저장 (ItemProcessor)
  • 만약 ItemProcessor가 존재하지 않는다면, 바로 Chunk<O>에 저장
• 외부로부터 ChunkProcessor가 호출될 때마다 항상 새로운 Chunk가 생성됨
• ItemProcessor 처리가 완료되면 Chunk<O>에 있는 List을 ItemWriter에게 전달
• ItemWriter 처리가 완료되면 Chunk 트랜잭션 종료 & Step 반복문에서는 다시 ChunkOrientedTasklet 새롭게 실행

[0JUUU]

ItemReader / ItemWriter / ItemProcessor 이해

ItemReader

• 다양한 입력으로부터 데이터를 읽어서 제공하는 인터페이스
• ChunkOrientedTasklet 실행 시 필수적 요소로 설정해야 함

• T read()
  • 입력 데이터를 읽고 다음 데이터로 이동
  • 아이템 하나를 리턴하며 더 이상 아이템이 없는 경우 null 리턴
  • 아이템 하나는 파일의 한줄, DB의 한 row 혹은 XML 파일에서 하나의 엘리먼트가 될 수 있음
  • 더 이상 처리해야 할 Item이 없어도 예외가 발생하지 않고 ItemProcessor와 같은 다음 단계로 넘아감
• 다수의 구현체들이 ItemReader와 ItemStream 인터페이스를 동시에 구현
  • ItemStream : 파일 스트림 연결 or 종료, DB 커넥션 연결 or 종료, 입력 장치 초기화 등의 작업
  • ExecutionContext에 read와 관련된 여러가지 상태 정보 저장 ➡️ 재시작 시 다시 참조하도록 지원

ItemWriter

• Chunk 단위로 데이터를 받아 일괄 출력 작업을 하는 인터페이스
• 아이템 하나가 아닌 아이템 리스트를 전달받음
• ChunkOrientedTasklet 실행 시 필수적 요소로 설정해야 함

• void write(List<? extends T> items)
  • 출력 데이터를 아이템 리스트로 받아 처리
  • 출력이 완료되고 트랜잭션이 종료되면 새로운 Chunk 단위 프로세스로 이동
• 다수의 구현체들이 ItemWriter와 ItemStream 인터페이스를 동시에 구현

ItemProcessor

• 데이터를 출력하기 전에 데이터를 가공, 변형, 필터링하는 역할
• ItemReader, ItemWriter와 분리되어 비즈니스 로직을 구현할 수 ⭕️
• ItemReader로부터 받은 아이템을 특정 타입으로 변환해서 ItemWriter에 넘겨줄 수 ⭕️
• ItemReader로부터 받은 아이템들 중 필터 과정을 거쳐 원하는 아이템들만 ItemWriter에게 넘겨줄 수 ⭕️
• ChunkOrientedTasklet 실행 시 선택적 요소 ➡️ Chunk 기반 프로세싱에서 ItemProcessor 단계가 반드시 필요한 것 ❌

• O process
  • <I> 제네릭은 ItemReader에서 받을 데이터 타입
  • <O> 제네릭은 ItemWriter에게 보낼 데이터 타입
  • 아이템 하나씩 가공 처리 ➡️ null 리턴할 경우, 해당 아이템은 Chunk<O>에 저장되지 ❌
• ItemStream을 구현하지 ❌ (ItemProcessor는 비즈니스 로직을 처리하는데에 사용하기 때문)
• 거의 대부분 Customizing해서 사용하기에 기본적으로 제공되는 구현체가 적음

[0JUUU]

ItemStream

• ItemReader와 ItemWriter 처리 과정 중 상태를 저장하고 오류가 발생하면 해당 상태를 참조하여 실패한 곳에서 재시작하도록 지원
• 리소스를 열고 (open) 닫아야(close)하며 입출력 장치 초기화 등의 작업을 해야하는 경우
• ExecutionContext를 매개변수로 받아 상태 정보를 업데이트(update) 함
• ItemReader 및 ItemWriter는 ItemStream을 구현해야 함

public class CustomItemStreamReader implements ItemStreamReader<String> {

    private final List<String> items;
    private int index = -1;
    private boolean restart = false;

    public CustomItemStreamReader(List<String> items) {
        this.items = new ArrayList<>(items);
        this.index = 0;
    }

    @Override
    public String read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
        String item = null;
        if (index < items.size()) {
            item = items.get(index);
            index++;
        }
        if (index == 6 && !restart) {
            throw new RuntimeException("Restart is required");
        }

        return item;
    }

    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {
        if (executionContext.containsKey("index")) {
            index = executionContext.getInt("index");
            restart = true;
        } else {
            index = 0;
            executionContext.put("index", index);
        }
    }

    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {
        executionContext.put("index", index);
    }

    @Override
    public void close() throws ItemStreamException {
        // 리소스 해제, 초기화 작업했던 것들을 해제
        System.out.println("close");
    }
}

1번째 실행

2번째 실행 ➡️ 실패한 이후부터 읽는다.

[0JUUU]

Chunk Process 아키텍처

1. Job을 실행하면 TaskletStep이 실행됨
2. TaskletStep은 내부에 RepeatTemplate이라는 반복기를 가지고 있어 ChunkOrientedTasklet을 반복
3. ChunkOrientedTasklet이 실행될 때 스프링 배치는 Transaction 경계 생성
4. Chunk 단위로 작업 시작
5. SimpleChunkProvider도 내부적으로 RepeatTemplate 반복기를 갖고 있어 ItemReader을 Chunk Size 만큼 반복시켜 데이터를 읽음
6. Chunk Size만큼 읽고, 읽은 아이템이 담긴 Chunk<I>를 SimpleChunkProcessor에 넘김
7. SimpleChunkProcessor는 전달받은 Chunk 데이터를 한 건씩 읽어 ItemProcessor로 데이터를 가공하여 Chunk<O>에 저장
8. ItemWriter에게 Chunk가 갖고 있는 List 값을 전달하고 ItemWriter는 출력 처리 (트랜잭션 커밋)
9. 이 과정이 Chunk 단위로 반복 , ItemReader에서 null 값을 읽을 때 반복작업 종료
   • 만약 중간에 예외 발생, 트랜잭션 롤백 & 작업 중단
   • 만약 ItemReader에서 null값을 읽어오게 된다면 RepeatStatus.FINISHED 를 통해 현재 작업을 마지막으로 다음부터는 반복 작업 ❌
   • Chunk 단위마다 새로운 트랜잭션 생성 / 커밋하는 과정이 존재