[Spring Batch] ItemProcessor 동작 원리와 4대 데이터 처리 전략

1. 도입부

웹 애플리케이션 개발에 익숙한 사람들에게 "데이터를 가져와서 가공한 뒤 저장한다"는 로직은 지극히 당연하고 단순한 흐름이다. 보통은 하나의 서비스 레이어에서 이 모든 과정이 이루어지기 때문이다. 하지만 수백만 건 이상의 대용량 데이터를 다루는 배치(Batch)의 세계로 들어오면 이야기가 완전히 달라진다.

스프링 배치(Spring Batch)는 이 흐름을 대용량 처리에 최적화된 구조로 쪼개어 제공한다. 바로 읽기(ItemReader), 처리(ItemProcessor), 쓰기(ItemWriter)로 역할을 엄격하게 분리하는 것이다. 이 중 ItemProcessor는 우리가 해결해야 하는 고유한 비즈니스 로직이 살아 숨 쉬는 가장 중요한 구간이다.

스프링 배치가 다양한 데이터베이스와 파일에 맞춤형 Reader와 Writer 구현체를 기본으로 제공하는 것과 달리, ItemProcessor는 표준 구현체가 거의 없다. 비즈니스 요구사항은 시스템마다 완전히 다르기 때문에 우리가 직접 규칙을 정의해야 한다. 하지만 자유도가 높다고 해서 아무렇게나 작성해도 된다는 뜻은 아니다. 내부 메커니즘과 상태 흐름을 오해하고 구현하면, 배치가 도중에 실패하거나 시스템 전체에 심각한 성능 장애를 초래할 수 있다.

이 글에서는 ItemProcessor가 동작하는 기본 원리부터 시작해서 데이터를 필터링, 검증, 변환, 보강하는 4가지 구체적인 방법과 성능을 100배 이상 끌어올릴 수 있는 실무 최적화 비결을 정리해 본다.

[Spring Batch] ItemStream의 자원 관리와 상태 복구

 

 

2. 주요 특징 및 핵심 로직

2.1 ItemProcessor의 정체와 청크 지향 단계의 결합

ItemProcessor를 이해하기 위해 가장 먼저 바라보아야 하는 것은 기본 인터페이스 명세다. 자바 8에 도입된 함수형 인터페이스(@FunctionalInterface)로 선언되어 있어 매우 명확하고 깔끔하다.

package org.springframework.batch.item;

import org.springframework.lang.Nullable;

@FunctionalInterface
public interface ItemProcessor<I, O> {
    @Nullable
    O process(I item) throws Exception;
}

제네릭 타입 파라미터인 <I, O>는 입력 타입(Input)과 출력 타입(Output)을 의미한다. 여기서 가장 중요한 포인트는 파라미터로 들어오는 item이 리스트나 묶음이 아니라 단 한 건의 아이템이라는 점이다.

스프링 배치의 청크 지향 단계(Chunk-oriented Step)는 아래와 같이 정해진 톱니바퀴처럼 반복적으로 맞물려 실행된다.

[데이터 읽기] -> Reader가 지정된 청크 크기(예: 10건)만큼 데이터를 로드하여 입력 청크 구성
                     ↓
[데이터 처리] -> 입력 청크의 아이템을 하나씩 순회하며 ItemProcessor.process() 호출 (10번 반복)
                     ↓
[데이터 쓰기] -> 가공이 끝난 아이템들을 다시 하나의 청크로 묶어 Writer.write()로 한 번에 전송

즉, process() 메서드는 청크 크기만큼 루프를 돌며 개별 건 단위로 수행된다. 이 작은 루프 단위의 실행 방식이 나중에 설명할 외부 통신 성능 함정의 원인이 되므로, 이 메커니즘을 머릿속에 정확히 각인해 두어야 한다.

 

 

3. 상세 가이드 및 심층 분석

ItemProcessor가 비즈니스 전장에서 데이터를 다루는 방식은 크게 4가지 전략으로 나뉜다. 각각의 동작 방식과 실제 코드를 함께 살펴보자.

3.1 전략 1: null 반환을 통한 데이터 필터링 (Filtering)

ItemProcessor에서 사용할 수 있는 가장 강력하면서도 안전한 제어권 중 하나는 바로 데이터 걸러내기다. process() 메서드를 실행하다가 특정 조건에 걸려 null을 반환하면, 스프링 배치는 이 아이템을 최종적으로 저장할 대상(Processed Chunk)에서 완전히 제외한다.

 

실제 필터링 구현 코드

아래 예제는 시스템에 침투하는 위험한 명령어 로그를 감시하고, 파괴적이거나 부적절한 권한 남용 행위가 포착되면 null을 반환하여 최종 적재 대상에서 말살하는 필터링 프로세서다.

import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.item.ItemProcessor;

@Slf4j
public class ExecutionerProcessor implements ItemProcessor<Command, Command> {

    @Override
    public Command process(Command command) throws Exception {
        // 1. 시스템을 파괴하는 명령어 감지 시 차단 및 null 반환
        if (command.getCommandText().contains("rm -rf /") ||
            command.getCommandText().contains("kill -9")) {
            log.info("사용자 {}의 위험 명령어 '{}' 포착 -> 기록에서 말살합니다.",
                    command.getUserId(), command.getCommandText());
            return null; // null 반환 시 최종 Chunk에서 조용히 제외됨
        }

        // 2. 권한을 불법적으로 남용하려는 의도가 포착되면 제외
        if (command.isSudoUsed() && command.getTargetProcess().contains("system")) {
            log.info("사용자 {}의 시스템 권한 남용 '{}' 포착 -> 기록에서 제외합니다.",
                    command.getUserId(), command.getCommandText());
            return null;
        }

        log.info("사용자 {}의 명령어 '{}' 검증 완료 -> 최종 기록 허가.",
                command.getUserId(), command.getCommandText());
        return command; // 유효한 아이템은 그대로 반환하여 다음 단계로 이동
    }
}

💡 주의: 모든 아이템이 걸러져 청크가 텅 비어버리면?

예를 들어 청크 크기가 10인데, 입력된 10개의 데이터가 전부 위험 명령어로 판정받아 모두 null을 반환했다면 배치는 멈출까? 멈추지 않는다. 스프링 배치는 빈 상태의 청크 객체를 들고 ItemWriter를 호출한다. 즉, 내용물만 비어 있을 뿐 ItemWriter.write()와 쓰기 앞뒤 단계의 리스너인 beforeWrite() 등은 정상적으로 가동된다. 따라서 쓰기 레이어를 직접 구현할 때는 청크 리스트가 비어 있는 상황에 대한 방어적인 예외 처리를 고려해 두어야 한다.

 

3.2 전략 2: ValidatingItemProcessor를 활용한 선언적 검증 (Validation)

스프링 배치는 데이터를 규칙에 따라 검사하고 걸러내는 빈번한 패턴을 위해 ValidatingItemProcessor라는 표준 컴포넌트를 미리 준비해 두었다. 이 클래스는 전용 검증 도구인 Validator 인터페이스와 결합하여 깔끔한 결합 분리를 이뤄낸다.

import org.springframework.batch.item.validator.ValidationException;
import org.springframework.batch.item.validator.Validator;

// 검증 책임만 고립시킨 유효성 검사기 구현
public class CommandValidator implements Validator<Command> {

    @Override
    public void validate(Command command) throws ValidationException {
        if (command.getCommandText().contains("rm -rf /") ||
            command.getCommandText().contains("kill -9")) {
            throw new ValidationException("치명적인 파괴 명령어 감지: " + command.getCommandText());
        }

        if (command.isSudoUsed() && command.getTargetProcess().contains("system")) {
            throw new ValidationException("비인가 권한의 시스템 제어 시도 감지");
        }
    }
}

이 검증 로직을 배치의 스텝 설정 파일(Configuration)에 적용할 때는 아래와 같이 두 가지 분기점을 선택할 수 있다.

@Bean
public ItemProcessor<Command, Command> commandProcessor() {
    ValidatingItemProcessor<Command> processor = new ValidatingItemProcessor<>(new CommandValidator());
    
    // 이 설정값에 따라 배치의 생사가 결정된다
    processor.setFilter(true); 
    
    return processor;
}

• setFilter(true): 검증 오류(ValidationException)가 발생하는 순간, 배치를 터트리는 대신 예외를 품어 안고 null을 조용히 반환한다. 즉, 오염된 데이터만 걸러내고 나머지는 끝까지 완수하는 유연한 대처다.
• setFilter(false) (기본값): 예외가 발생하는 즉시 트랜잭션과 단계를 중단하고 전체 배치를 에러 상태(FAILED)로 완전 종료한다. 단 한 건의 오염도 허용하지 않는 단호한 실패 처리 기법이다.

 

3.3 전략 3: 데이터 변환 (Transformation)

ItemProcessor의 가장 원초적이고 대표적인 쓰임새는 읽어온 데이터 원본의 타입과 구성을 아예 다른 형태의 출력용 모델로 환골탈태시키는 데이터 변환(Transformation)이다.

예를 들어, 데이터베이스에서 가져온 복잡한 원시 시스템 로그(SystemLog) 객체를 기반으로, 정합성 분석 연산을 수행해 정형화된 위험 레벨 분석 리포트 객체(CommandReport)로 변환하는 시나리오를 구성할 수 있다.

// 읽기 대상 원본 모델
@Data
public class SystemLog {
    private Long userId;          // 실행한 행위자 ID
    private String rawCommand;    // 실행된 원시 텍스트 명령어
    private LocalDateTime executedAt; // 실행 일시
}

// 쓰기 대상 타겟 모델
@Data
public class CommandReport {
    private Long executorId;      // 변환된 행위자 ID
    private String action;        // 분석 요약 행동 메시지
    private String severity;      // 비즈니스 위험 등급 (CRITICAL, HIGH, LOW)
    private LocalDateTime timestamp; // 기록 시각
}

이 두 가지 서로 다른 도메인 타입을 징검다리처럼 잇는 변환기 역할을 ItemProcessor가 맡게 된다.

import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.item.ItemProcessor;

@Slf4j
public class CommandAnalyzer implements ItemProcessor<SystemLog, CommandReport> {

    @Override
    public CommandReport process(SystemLog systemLog) throws Exception {
        // 입력은 SystemLog 타입이지만 반환은 CommandReport 타입으로 창조
        CommandReport report = new CommandReport();
        report.setExecutorId(systemLog.getUserId());
        report.setTimestamp(systemLog.getExecutedAt());

        String command = systemLog.getRawCommand();
        if (command.contains("rm -rf")) {
            report.setAction("스토리지를 파괴하는 비정상 명령어 격리");
            report.setSeverity("CRITICAL");
        } else if (command.contains("kill -9")) {
            report.setAction("운영 데몬 강제 종료 추적");
            report.setSeverity("HIGH");
        } else {
            report.setAction("정상 가동 범위의 유틸리티 실행");
            report.setSeverity("LOW");
        }

        log.info("로그 가공 완료: {} -> 등급: {}", systemLog.getUserId(), report.getSeverity());
        return report;
    }
}

이 구조를 사용하면 Reader는 순수하게 인풋 엔티티만 바라보고, Writer는 가공이 완료된 전용 엔티티나 리포트 객체만 바라볼 수 있게 되므로 아키텍처 결합도가 크게 낮아진다.

 

3.4 전략 4: 데이터 보강 (Data Enrichment)

때로는 읽어온 데이터 원본에 뼈대만 있고 살이 없는 경우가 있다. 예를 들어 사용자 로그 데이터에는 ID만 찍혀 있고 구체적인 '서버명'이나 '소속 정보'가 비어 있어서, 외부 관측 API나 원격 클라이언트를 호출해 빈 곳을 든든하게 메꿔 넣어야 하는 상황이다. 이를 데이터 보강(Data Enrichment)이라 부른다.

• 변환과 보강의 결정적 차이점:
  • 변환: 객체의 외형 자체를 혁신하는 작업 (SystemLog -> CommandReport)
  • 보강: 원본 객체의 타입은 그대로 둔 채, 비어 있는 멤버 변수에 외부 정보를 밀어 넣어 풍성하게 만드는 작업 (riskLevel = null -> riskLevel = HIGH)

 

 

4. 실무 팁 및 주의사항 (외부 시스템 통신 최적화)

실무에서 데이터 보강 전략을 구현할 때, 주니어 개발자들이 가장 자주 실수를 저지르고 시스템을 마비시키는 성능 파괴의 폭탄이 바로 여기에 숨겨져 있다.

⚠️ 개별 API 호출의 끔찍한 성능 참사

앞서 강조했듯이 ItemProcessor.process()는 개별 데이터 한 건마다 한 번씩 루프를 돌며 호출된다. 만약 100만 건의 방대한 데이터를 처리하는 배치 작업이 있고, 프로세서 내부에서 건당 정보를 보강하기 위해 매번 외부 모니터링 시스템의 API를 호출하도록 설계했다고 치자.

만약 외부 서버와 통신하는 데 물리적인 지연 시간을 포함하여 건당 평균 100밀리초(0.1초)의 시간이 걸린다고 가정해 보자.

• 1건 처리 시간: 100밀리초 (0.1초)
• 100만 건 개별 호출 대기 시간: 총 10만 초

이 10만 초를 시간 단위로 풀어보면 무려 약 27.7시간이라는 무지막지한 대기 지연이 발생한다. 하루가 꼬박 넘는 시간 동안 배치 서버는 네트워크 통신 대기 상태로 놀고 있으며, 서비스를 받아 주는 타겟 API 서버 또한 끝도 없이 들어오는 단발성 트래픽 공격에 서버가 마비될 위험에 처한다.

 

💡 해결책: ItemWriteListener의 벌크(Bulk) 처리 최적화

이 병목 현상을 타파하는 우아한 해법은 바로 벌크(Bulk) API와 스프링 배치의 리스너 연계에 있다.

스프링 배치가 지원하는 ItemWriteListener의 beforeWrite 단계는 단일 데이터가 아니라 쓰기가 예정된 가공 완료 청크 묶음 전체(List)를 파라미터로 제공한다.

 

벌크 조회 리스너 구현 예제

이 기법을 사용하면 청크에 모인 여러 식별자들을 한 번에 리스트로 추출한 뒤, 단 한 번의 벌크 API 연동 통신으로 정보를 통째로 수집해 보강을 끝마칠 수 있다.

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.ItemWriteListener;
import org.springframework.batch.item.Chunk;
import java.util.List;
import java.util.Map;

@Slf4j
@RequiredArgsConstructor
public class SystemLogEnrichListener implements ItemWriteListener<SystemLog> {

    // 외부 연동용 통신 서버 모듈 주입
    private final ObservabilityApiClient observabilityApiClient;

    @Override
    public void beforeWrite(Chunk<? extends SystemLog> items) {
        // 1. 청크에 보관된 사용자 ID 전체를 일괄 수집하여 리스트로 묶음
        List<Long> userIds = items.getItems().stream()
                .map(SystemLog::getUserId)
                .toList();

        // 2. 단 한 번의 네트워크 API 호출로 청크 단위 일괄 정보 획득 (통신 비용 절감)
        Map<Long, ServerInfo> serverInfoMap = observabilityApiClient.fetchServerInfos(userIds);

        // 3. 응답 맵 데이터를 기존 수집 데이터에 부드럽게 결합하여 보강 완료
        items.getItems().forEach(systemLog -> {
            ServerInfo serverInfo = serverInfoMap.get(systemLog.getUserId());
            if (serverInfo != null) {
                systemLog.setServerName(serverInfo.getHostName());
                systemLog.setProcessName(serverInfo.getCurrentProcess());
                systemLog.setRiskLevel("NORMAL");
            }
        });

        log.info("성능 최적화 완료: 벌크 통신을 활용하여 {}건의 정보 일괄 보강 완료.", items.size());
    }
}

최적화 결과 수치 비교

배치의 청크 크기를 100으로 넉넉히 지정해 두었다면, 100만 건의 데이터를 처리하기 위해 필요한 전체 네트워크 통신 횟수는 100만 번에서 1/100 비율인 단 1만 번으로 획기적으로 줄어든다.

평균 통신 시간 100밀리초 기준 단 1만 번의 대기 시간만 누적되면 되므로, 전체 대기 지연 시간은 총 1000초가 되어 단 16분밖에 소요되지 않는다.

동일한 양의 데이터를 똑같은 방식으로 보강했음에도, 아키텍처 설계 하나 바꾼 덕분에 대기 시간이 27.7시간에서 16분으로 줄어드는 약 100배 이상의 압도적인 처리 속도 향상을 확보하게 된다.

 

❓ 쓰면서 조회하면 되지, 왜 귀찮게 리스너를 분리할까?

"어차피 일괄로 쓰기가 동작하는 ItemWriter.write(List) 메서드 내부에서 벌크로 API를 부르고 곧바로 로컬 디스크에 기록하면 훨씬 간결하지 않을까?"라는 의문이 생길 수 있다.

하지만 이는 자바 엔터프라이즈의 핵심 가치인 단일 책임 원칙(Single Responsibility Principle)을 뒤흔드는 유해한 결합이다. 스프링 배치 프레임워크가 정한 각 구간의 수호 역할은 명확해야 한다.

• ItemProcessor / Listener: 데이터의 가공, 정제, 정합성 검사 및 보강
• ItemWriter: 가공이 완료되어 완성형 상태로 들어온 리스트 데이터를 오직 영속 매체(DB, NoSQL 등)에 쓰기(Write) 연산만 집중 수행

만약 Writer 내부에 무관한 외부 API 조회용 원격 코드들이 질척하게 섞이게 된다면, 나중에 가혹한 가동 환경 변화로 인해 데이터베이스 적재 장소(RDB)를 통째로 NoSQL이나 외부 파일 전송으로 이관해야 할 때, 전혀 무관한 API 호출 네트워크 구성 코드까지 다시 복사해서 붙여넣어야 하는 비극이 일어난다. 컴포넌트의 가치가 독립적으로 살아남을 수 있도록 리스너에서 전처리 보강을 끊어 주는 것이 유연성 측면에서 훨씬 뛰어나다.

 

 

7. 복합 위임 프로세서의 정렬과 고급 활용

실무에서는 하나의 일차적인 연산만 거치는 가벼운 스텝보다 '1단계 필터링' -> '2단계 타입 변환' -> '3단계 보안 검사' 등 여러 개의 순차 프로세서를 결합해야 하는 상황이 빈번하다. 스프링 배치는 이 프로세서들을 파이프라인처럼 연결하고 분기하는 위임 프록시 도구를 지원한다.

CompositeItemProcessor (순차 연계 체인)

CompositeItemProcessor는 체인 리스트에 보관된 하위 대상 프로세서들을 정해진 순서대로 연속 통과시키는 파이프라인 전용 위임 프로세서다.

이 체인은 이전 단계의 반환값이 다음 단계 프로세서의 입력값으로 징검다리처럼 안전하게 호환 연결되어야 한다.

import org.springframework.batch.item.support.CompositeItemProcessor;
import java.util.List;

@Bean
public CompositeItemProcessor<SystemLog, CommandReport> compositeProcessor() {
    CompositeItemProcessor<SystemLog, CommandReport> compositeProcessor = new CompositeItemProcessor<>();
    
    // 순차 연속 연계 패턴 구성
    compositeProcessor.setDelegates(List.of(
            new SystemLogValidationFilter(), // 1단계: 유효성 필터링 (SystemLog -> SystemLog)
            new CommandAnalyzer()            // 2단계: 결과 분석 보고서 변환 (SystemLog -> CommandReport)
    ));
    
    return compositeProcessor;
}

이 패턴을 이용하면 작은 비즈니스 조각들을 독립된 클래스로 쪼개어 가독성을 높이고, 레고 블록을 맞추듯이 스텝 설정 파일에서 자유롭게 순서를 조합해 나갈 수 있다.

 

ClassifierCompositeItemProcessor (동적 다중 라우팅)

만약 들어오는 개별 레코드 정보의 특정 조건(예: 위험한 계정 여부)에 따라, 비즈니스 연산을 타는 대상 프로세서 객체 자체를 동적으로 분기시켜 타게 만들고 싶다면 ClassifierCompositeItemProcessor가 해답이 된다.

import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.support.ClassifierCompositeItemProcessor;
import org.springframework.classify.Classifier;

@Bean
public ClassifierCompositeItemProcessor<SystemLog, CommandReport> classifierProcessor() {
    ClassifierCompositeItemProcessor<SystemLog, CommandReport> classifierProcessor = new ClassifierCompositeItemProcessor<>();
    
    // 조건 매칭 라우팅 분류기 설정
    classifierProcessor.setClassifier(new Classifier<SystemLog, ItemProcessor<?, ? extends CommandReport>>() {
        @Override
        public ItemProcessor<?, ? extends CommandReport> classify(SystemLog classifiable) {
            // 특별 관리 대상인 슈퍼 관리자 계정 그룹일 경우 복잡 심층 분석기 배정
            if (classifiable.getUserId().equals(666L)) {
                return new HighSeverityCommandAnalyzer(); 
            }
            // 그 외 일반 계정 그룹은 표준 경량 분석기 객체 배정
            return new CommandAnalyzer();
        }
    });
    
    return classifierProcessor;
}

단, 이때 분류기가 리턴해 주는 다양한 커스텀 프로세서들의 최종 반환 타입은 상위 마스터인 ClassifierCompositeItemProcessor가 정의한 출력 타입과 반드시 논리적인 상속 및 호환 관계를 만족하고 있어야 타입 정렬 오류 없이 매끄럽게 컴파일된다는 사실을 명심해 두자.

 

 

8. 마무리

스프링 배치의 심장부인 ItemProcessor를 다루는 4가지 방법과 핵심 특징을 가볍게 되짚어 보자.

• 데이터 필터링: 불필요한 데이터를 가공 처리 단계에서 null을 반환해 솎아내며 최종 저장 페이로드를 줄인다.
• 검증 실패 처리: setFilter(false) 설정을 활용해 오염된 정산 데이터 유입 시 즉시 전체 프로세스를 폭파시키고 신속 경고를 날린다.
• 데이터 변환: 입력 타입과 출력 타입을 독립적으로 가져가 각 컴포넌트 간의 타입 지향 결합 강도를 유연하게 격리한다.
• 데이터 보강: 외부 API 조회가 필요할 때 프로세서 안에서 건당 부르는 무식한 방법을 버리고, ItemWriteListener의 벌크 통신 기법을 도입해 네트워크 지연 대기 비용을 수십 시간에서 단 몇 분 단위로 단축하는 기적적인 성능 최적화를 일궈낸다.

배치 아키텍처는 성능에 타협을 주지 않으면서도 각 컴포넌트의 가치를 유지하기 위한 치열한 구조적 분리 설계의 연속이다. 우리가 정의한 고유한 비즈니스 로직들이 대용량 트래픽의 파도 속에서도 안정적으로 지탱될 수 있도록, 위에서 살펴본 벌크 연계 패턴과 복합 프록시 구조를 적재적소에 녹여내어 견고한 데이터 마스터 환경을 이룩해 보길 바란다.