ItemProcessor

우리의 비즈니스 로직이 살아 숨쉬는 전장으로 진입해보자.
이제부터 ItemProcessor를 올바르게 다루는 방법을 파헤쳐보자.

첫 번째 방식: null 반환을 통한 데이터 필터링

ItemProcessor의 process() 메서드가 null을 반환하면 해당 item은 ItemWriter로 전달되는 Chunk에서 완전히 제외된다.

• Spring Batch는 먼저 지정된 청크 사이즈만큼 read() 메서드를 호출하여 input Chunk를 생성한다.
• 생성된 input Chunk의 각 item에 대해 ItemProcessor의 process() 메서드를 호출한다.
• 이 과정에서 process()가 null을 반환한 item은 최종 output Chunk에서 제외된다.
• 결과적으로 ItemWriter에 전달되는 output Chunk의 크기는 input Chunk의 크기보다 작아진다.

대표적인 필터링 시나리오는 다음과 같다

• 유효하지 않은 데이터 제거 (비정상적인 금액, 잘못된 주문상태 등)
• 처리가 불필요한 데이터 제외 (휴면계정, 탈퇴회원 등)
• 특정 조건에 맞지 않는 데이터 제외 (기준금액 이하 거래, 특정 상태의 주문 등)

Spring Batch에서는 이러한 null 반환 방식을 활용해 데이터를 검증하는 ItemProcessor 구현체를 제공한다. 바로 ValidatingItemProcessor 이다.

ValidatingItemProcessor는 org.springframework.batch.item.validator.Validator를 사용해 데이터 필터링을 수행한다. Validator 인터페이스는 다음과 같다.

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

사용법은 다음과 같다.

@Bean
piublic ItemProcessor<Item, Item> itemProcessor() {
    ValidatorItemProcessor<Item> processor = new ValidatingItemProcessor<>(validator);
    processor.setFilter(true); // ValidationException 발생 시 필터링 수행
    return processor;
}

• ValidatorItemProcessor로 데이터 필터링을 수행하기 위해서는 setFilter(true)로 설정해야 한다.
• 이렇게 설정하면 Validator에서 ValidationException이 발생했을 떄 해당 데이터가 필터링되도록 null이 반환된다.
• 기본값인 setFilter(false)로 설정된 경우에는 ValidationException이 발생하면 예외가 상위로 전달되어 Step이 실패하게 된다.
  • 예외가 상위로 전파되면, 현재 Step은 물론 전체 Job까지 바로 실패해버린다.
  • 이것이 바로 데이터 검증을 통한 실패 처리 방식이다.
  • 유효하지 않는 데이터 하나만 발견되어도 즉시 예외를 던져 전체 배치 잡을 중단시킬 수 있다.

두 번째 방식: 데이터 변환

ItemProcessor의 가장 핵심적인 역할은 읽어온 데이터를 우리가 원하는 형태로 변환하는 것이다.
비즈니스 요구사항에 맞춰 데이터를 변환하는 것이 ItemProcessor의 존재 이유다.

세 번째 방식: 데이터 보강

때로는 읽어온 데이터만으로 충분하지 않을때가 있다. 외부 시스템이나 데이터베이스에서 추가 정보를 가져와 기존 데이터를 보강해야 하는 경우가 있다.

이것이 바로 데이터 보강(Data Enrichment)이다. 원본 데이터에 외부 소스로 얻은 정보를 추가하여 더욱 풍부하고 의미 있는 데이터로 만드는 과정이다.

대표적인 데이터 보강 시나리오는 다음과 같다.

• 거래 내역에 실시간 환율 적용 (외환 API를 통한 원화 환산)
• 주문 데이터에 재고 현황 추가 (창고 시스템 API 조회)
• IP 주소에 지역 정보 보강 (GeoIP API를 통한 국가/도시 정보)

그러나 데이터 보강 이면에는 놓치면 안 되는 치명적인 위험이 있다.

외부 시스템 통신 최적화

ItemProcessor의 process() 메서드는 아이템을 하나씩 처리하는 단위성 작업이다.
이런 특성으로 인해 데이터 보강이나 외부 API나 DB 조회가 필요한 경우 심각한 성능 이슈가 발생할 수 있다.

이 한계를 어떻게 돌파할 수 있을까? ItemWriteListener를 활용해볼 수 있을 것이다.
1장에서 살펴본 ItemWriteListener의 beforeWrite() 메서드 시그니처를 다시 한 번 살펴보자.

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

beforeWrite() 메서드의 파라미터 타입에 주목해보자. 개별 item이 아닌 Chunk 전체를 입력으로 받는 것을 알 수 있다.
외부 API가 지원한다면, 여러 건을 동시에 조회하는 API를 사용해서 통신 횟수를 줄여볼 수 있을 것이다.

ItemWriter.write() 에서 API를 호출하고 바로 쓰기까지 한번에 처리하면 안되나?
ItemWriter는 읽기 작업을 설계된 컴포넌트가 아니다.
단일 책임 원칙 (Single Responsibility Principle) 위반이다. 따라서 이런 접근 시도는 하지말자.

필터링, 검증, 변환, 보강까지 ItemProcessor의 역할을 모두 살펴봤다.
이제 한 단계 더 올라가 ItemProcessor들을 조합하고 연계하는 방법을 살펴보자.

CompositeItemProcessor

CompositeItemProcessor는 여러 위임 대상 ItemProcessor를 순차적으로 실행하는 위임 ItemProcessor 구현체다. 각 ItemProcessor는 순차적으로 실행되며, 이전 ItemProcessor의 반환 값이 다음 ItemProcessor의 입력으로 전달된다.

이때 타입의 연속성이 매우 중요한데, 첫 번쨰 ItemProcessor의 입력 타입은 ItemReader의 반환 타입과 일치해야 하고,
마지막 ItemProcessor의 반환 타입은 ItemWriter가 처리할 수 있는 타입이어야 한다.
또한 중간에 위치한 ItemProcessor들도 서로 입력과 출력 타입이 일치해야 한다.

CompositeItemProcessor를 사용하기 위해서는 생성자에 위임 대상 ItemProcessor들을 전달하거나, CompositeItemProcessorBuilder.delegates() 메서드를 통해 설정하면 된다.

다음 작전에서는 청크 지향 처리를 더욱 견고하게 구축하는 기술울 파헤쳐보도록 하자.