TIL

ItemProcessor

• ItemProcessor에는 고유한 비즈니스 로직이 들어가는 곳이다.
  • 그래서 Spring이 제공하는 별도의 구현체는 거의 없다.

null 반환을 통한 데이터 필터링

• ItemProcessor의 process() 메서드가 null을 반환하면 해당 item은 제외된다.
• 진행 과정
  • 청크 사이즈만큼 read() 메서드를 호출해 청크를 생성
  • 청크의 각 item에 대해 process() 메서드 호출
  • process()에서 null이 반환되면 최종 청크에서 제외
  • ItemWriter에 전달되는 청크 크기는 초반 청크의 크기보다 작아진다.
• 보통 null이 반환되는 경우
  • 유효하지 않은 데이터
  • 처리가 불필요한 데이터
  • 특정 조건에 맞지 않는 데이터

ValidatingItemProcessor

• Spring Batch에서 제공하는 null 반환 방식의 데이터 검증을 수행하는 ItemProcessor 구현체
• ValidatingItemProcessor를 생성하려면 Validator 인터페이스가 필요하다.
  • org.springframework.batch.item.validator.Validator

public interface Validator<T> {
    void validate(T value) throws ValidationException;
}

public class CommandValidator implements Validator<Command> {
    @Override
    public void validate(Command command) throws ValidationException {
        // ... 특정 로직에 따라 ValidationException 던지기
    }
}

• T 제네릭 타입(item의 타입)을 받아 ValidationException을 던지도록 구현하면 된다.
  • 이 때 setFilter(true)를 설정해야 ValidationException이 발생했을 때 null이 반환된다.
  • setFilter(false)가 되면 예외가 상위로 전달되어 Step이 실패한다.

@Bean
public ItemProcessor<Command, Command> commandProcessor() {
    ValidatingItemProcessor<Command> processor = 
        new ValidatingItemProcessor<>(new CommandValidator());
    processor.setFilter(true);  // ValidationException 발생 시 필터링 수행
    return processor;
}

데이터 검증을 통한 실패 처리

• ValidatingItemProcessor를 setFilter(false)로 설정하면 예외가 상위로 전파된다.
• 이는 전체 배치 잡을 중단시키게 된다.
  • Step 실패 → Job 실패로 전파
  • 뿐만 아니라 process() 메서드에서 예외를 던지면 동일한 결과를 얻는다.

데이터 변환

• ItemProcessor의 가장 핵심적인 역할은 데이터를 다른 데이터로 변환하는 것이다.
• ItemReader에서 읽은 데이터는 그대로는 사용하기 어려운 경우가 많다.
  • ex) 레거시 시스템 데이터를 신규 포멧으로 변환
• ItemProcessor의 제네릭 타입
  • I : ItemReader로부터 읽은 데이터 타입
  • O : ItemWriter로 전달할 데이터 타입
• 이 제네릭 타입은 chunk() 메서드의 제네릭 타입과 연결된다.
  • .<Input, Output>chunk()

데이터 보강

• 때로는 읽어온 데이터만으로 충분하지 않은 경우가 있다.
• 데이터 변환 vs 데이터 보강
  • 변환: 원본 타입을 다른 타입으로 변경
  • 보강: 출력 데이터에 부가 정보를 추가하는 것
• 대표적인 데이터 보강 시나리오
  • 거래 내역에 실시간 환율 적용
  • 주문 데이터에 재고 현황 추가
  • IP 주소에 지역 정보 보강

외부 시스템 통신 최적화

• process() 메서드는 아이템을 하나씩 처리하는 작업이기에 데이터 보강 때문에 외부 api나 db 조회가 필요한 경우 성능 이슈가 발생할 수 있다.
• ItemWriterListener의 beforeWrite()를 사용해 이 문제를 개선할 수 있다.
  • 아래 파라미터 타입을 보면 Chunk 전체를 입력 받는다.
  • 전체 청크에 대해 데이터 보강을 최적화할 수 있는 것이다.
  • ex) DB 조회 시 IN 절로 조회 최적화

default void beforeWrite(Chunk<? extends S> items) {
}

CompositeItemProcessor

• 여러 위임 대상 ItemProcessor를 순차적으로 실행하는 구현체
• 이전 ItemProcessor의 반환 값이 다음 ItemProcessor의 입력으로 전달된다.
  • 즉 첫 번째 ItemProcessor 입력 타입은 ItemReader의 반환 타입과, 마지막 ItemProcessor 반환 타입은 ItemWriter의 입력 타입이어야 한다.
• 생성자에 위임대상 ItemProcessor들을 전달하거나 CompositeItemProcessorBuilder.delegates()메서드를 사용

public CompositeItemProcessor(ItemProcessor<?, ?>... delegates) {
    this(Arrays.asList(delegates));
}

public CompositeItemProcessorBuilder<I, O> delegates(List<? extends ItemProcessor<?, ?>> delegates) {
    this.delegates = delegates;

    return this;
}

ClassifierCompositeItemProcessor

• Classifier를 사용해 아이템을 처리할 ItemProcessor를 라우팅하는 구현체
  • Classifier는 입력 아이템을 받아서, 타입 선언에 맞는 ItemProcessor를 반환

// 생성자 사용
public void setClassifier(Classifier<? super I, ItemProcessor<?, ? extends O>> classifier) {
    this.classifier = classifier;
}

// ClassifierCompositeItemProcessorBuilder 사용
public ClassifierCompositeItemProcessorBuilder<I, O> classifier(
        Classifier<? super I, ItemProcessor<?, ? extends O>> classifier) {
    this.classifier = classifier;
    return this;
}

• Classifier가 반환하는 모든 ItemProcessor의 입/출력 타입은 ClassifierCompositeItemProcessor에서 선언된 타입과 일치해야 한다