[Spring Batch 5] API 호출 Item Reader 커스텀과 csv 데이터 생성

이 글에서는 Spring Batch에서 외부 API 호출 결과를 처리하는 커스텀 PagingItemReader와 CSV 데이터를 생성하는 ItemWriter의 구현 방법을 설명합니다. Spring Batch 공식 문서를 기준으로 각 컴포넌트의 역할과 설계 의도를 기술합니다.

소개 · 배경

Spring Batch는 대용량 데이터를 일괄 처리하기 위한 프레임워크입니다. Spring Batch Reference Manual에 따르면, 배치 작업은 데이터를 읽고(Read), 처리하고(Process), 쓰는(Write) 단계로 구성됩니다. 이러한 처리 모델을 Chunk-oriented Processing이라고 합니다.

 

실무에서는 RDBMS뿐만 아니라 외부 API나 파일 시스템 등 다양한 데이터 소스를 처리해야 하는 경우가 있습니다. Spring Batch는 이러한 요구사항을 충족하기 위해 ItemReader, ItemProcessor, ItemWriter 인터페이스를 제공합니다. 각 인터페이스는 확장 가능하도록 설계되어 있어, 표준 구현체로 해결할 수 없는 경우 커스텀 구현을 통해 비즈니스 요구사항을 충족할 수 있습니다.

 

외부 API 기반 데이터 처리는 네트워크 지연, 페이지네이션, 에러 처리 등의 특수성을 가지고 있습니다. Spring Batch는 이러한 특수성을 고려하여 AbstractPagingItemReader와 AbstractItemCountingItemStreamItemReader 같은 추상 클래스를 제공합니다. 이러한 추상 클래스들은 페이징 기반 데이터 읽기를 일반화하여, 다양한 데이터 소스에 일관된 방식으로 접근할 수 있도록 설계되었습니다.

 

기본 개념 설명

Tasklet Job과 Chunk Job

Spring Batch는 두 가지 작업 모델을 제공합니다. Tasklet 모델은 단일 작업 단위를 실행하는 방식이며, Chunk 모델은 데이터를 청크 단위로 읽고 처리하는 방식입니다. Spring Batch Reference Manual에 따르면, 대용량 데이터 처리를 위해서는 Chunk 모델이 권장됩니다.

 

Chunk 모델은 ItemReader, ItemProcessor, ItemWriter를 조합하여 구성합니다. ItemReader는 데이터 소스로부터 아이템을 읽어오는 역할을 담당하며, ItemProcessor는 읽어온 데이터를 변환하거나 필터링하는 역할을 수행합니다. ItemWriter는 처리된 데이터를 목적지에 쓰는 역할을 담당합니다.

Item Reader, Item Processor, Item Writer 처리 흐름

Spring Batch Reference Manual에 정의된 Chunk 처리 흐름은 다음과 같습니다. ItemReader는 read() 메서드를 통해 아이템을 하나씩 반환합니다. 반환된 아이템은 ItemProcessor의 process() 메서드를 통해 변환되거나 필터링됩니다. 처리된 아이템들은 청크 크기만큼 누적된 후, ItemWriter의 write() 메서드를 통해 일괄 처리됩니다.

 

이러한 처리 흐름은 메모리 효율성과 트랜잭션 관리를 최적화하기 위해 설계되었습니다. 청크 단위로 처리함으로써 대용량 데이터도 제한된 메모리로 처리할 수 있으며, 각 청크는 독립적인 트랜잭션으로 실행되어 장애 격리가 가능합니다.

 

Custom ItemReader 파트

AbstractPagingItemReader

AbstractPagingItemReader는 Spring Batch가 제공하는 페이징 기반 데이터 읽기를 위한 추상 클래스입니다. Spring Batch API Javadoc에 따르면, 이 클래스는 AbstractItemCountingItemStreamItemReader를 상속받아 구현되었습니다.

 

AbstractPagingItemReader는 페이징 처리를 일반화하여, 다양한 데이터 소스에 일관된 방식으로 접근할 수 있도록 설계되었습니다. 이 클래스는 내부적으로 페이지 번호를 관리하고, doReadPage() 메서드를 호출하여 각 페이지의 데이터를 읽어옵니다. 읽어온 데이터는 내부 큐에 저장되며, doRead() 메서드를 통해 하나씩 반환됩니다.

 

외부 API 환경에서 이 추상 클래스를 사용하는 이유는 다음과 같습니다. 외부 API는 일반적으로 페이지네이션을 지원하며, 각 페이지를 독립적으로 요청할 수 있습니다. AbstractPagingItemReader는 이러한 특성을 활용하여 페이지 단위로 데이터를 읽어오는 패턴을 제공합니다. 또한 ItemStream 인터페이스를 구현하여 재시작 가능한 작업을 지원하므로, 장애 발생 시 이전 상태에서 재개할 수 있습니다.

AbstractItemCountingItemStreamItemReader

AbstractItemCountingItemStreamItemReader는 Spring Batch가 제공하는 아이템 카운팅 기반 스트림 읽기를 위한 추상 클래스입니다. Spring Batch API Javadoc에 따르면, 이 클래스는 ItemReader와 ItemStream 인터페이스를 모두 구현합니다.

 

ItemStream 인터페이스는 배치 작업의 상태를 저장하고 복원하는 기능을 제공합니다. open(), update(), close() 메서드를 통해 작업 실행 컨텍스트와 상호작용하여, 재시작 가능한 작업을 구현할 수 있습니다. AbstractItemCountingItemStreamItemReader는 이러한 기능을 기반으로 아이템 카운팅을 통해 현재 읽기 위치를 추적합니다.

 

ItemStream과 ItemReader의 결합 구조는 Spring Batch의 재시작 가능성(restartability)을 보장하기 위한 설계입니다. 작업 실행 중 장애가 발생하면, 저장된 실행 컨텍스트를 기반으로 마지막 처리 지점부터 재개할 수 있습니다. 외부 API 환경에서도 이러한 재시작 가능성은 중요합니다. 네트워크 오류나 API 제한으로 인한 장애 발생 시, 이미 처리한 데이터를 다시 읽지 않고 이어서 처리할 수 있습니다.

 

 

Spring Batch AbstractPagingItemReader를 사용한 커스텀 클래스 작성하기

 

 

커스텀 PagingItemReader 구현

실무에서 사용한 외부 API 호출을 위한 커스텀 PagingItemReader 구현체입니다.

 

@Slf4j
public class ###APIPageItemReader<T> extends AbstractPagingItemReader<T> {

    protected ###ApiFetcher ###ApiFetcher;
    protected LocalDate apiReleaseDate;

    public ###APIPageItemReader(int pageSize
                                    , ###ApiFetcher ###ApiFetcher
                                    , LocalDate apiReleaseDate) {
        setPageSize(pageSize); 
        this.###ApiFetcher = ###ApiFetcher; 
        this.apiReleaseDate = apiReleaseDate; 
    }


    @Override
    protected void doReadPage() { 
        initResults();

        /** ### API Call - 물류센터 WMS 등록된 당일 출고 데이터 조회 */
        List<ReleasedProductApiResponse> releasedProductsForThreeDays = getReleasedProductsForThreeDays(apiReleaseDate);

        /** 주문번호 필터링, 송장번호 필터링, 주문번호-송장번호 중복 제거 */
        List<ReleasedProductApiResponse> filtered = releasedProductsForThreeDays
            .stream()
            .filter(product -> !product.getSalesOrderNumber().startsWith("C") && !product.getSalesOrderNumber().startsWith("A"))
            .filter(product -> StringUtils.isNumberic(product.getTrackingNumber()))
            .distinct()
            .toList();

        int totalSize = filtered.size();

        int page = getPage() >= 0 ? getPage() : 1;
        int pageSize = getPageSize();
        int fromIndex = page * pageSize;
        int toIndex = Math.min(fromIndex + pageSize, totalSize);

        List<ReleasedProductApiResponse> pagedReleasedProductsForThreeDays
            = new ArrayList<>(filtered.subList(fromIndex, toIndex));

        for (ReleasedProductApiResponse releasedProductApiResponse : pagedReleasedProductsForThreeDays) {
            results.add((T) releasedProductApiResponse);
        }
    }

    protected void initResults() {
        if (CollectionUtils.isEmpty(results)) {
            results = new CopyOnWriteArrayList<>();
        } else {
            results.clear();
        }
    }

    /** releaseDate 당일치만 연동 */
    public List<ReleasedProductApiResponse> getReleasedProductsForThreeDays(LocalDate releaseDate) {
        PageResponse<ReleasedProductApiResponse> tempReleasedDay = ###ApiFetcher.getOrderReleaseBarcodeApiResponse(
            releaseDate, 10, 1);
        Long totalSize = tempReleasedDay.getTotalSize();
        PageResponse<ReleasedProductApiResponse> AllReleasedDayResponse = ###ApiFetcher.getOrderReleaseBarcodeApiResponse(
            releaseDate, totalSize.intValue(), 1);

        List<ReleasedProductApiResponse> releasedProducts = AllReleasedDayResponse.getList();

        return releasedProducts;
    }


    @Override
    protected void doOpen() throws Exception {
		... 
   }


    @Override
    protected void doClose() throws Exception {
        ... 
    }


}

 

구성요소	역할
setPageSize()	한 번 읽을 페이지 크기 설정
doReadPage()	외부 API 호출 + 결과 필터링 + 페이징 결과 추출
results	현재 페이지의 읽은 데이터 저장
initResults()	이전 결과를 초기화
getReleasedProductsForThreeDays()	외부 API 호출 및 전체 리스트 수집

 

 

---

 

 

ItemWriter 파트: FlatFileItemWriter를 사용해 csv 데이터 생성하기  

FlatFileItemWriter

FlatFileItemWriter는 Spring Batch가 제공하는 플랫 파일 쓰기를 위한 클래스입니다. Spring Batch API Javadoc에 따르면, 이 클래스는 Resource에 데이터를 쓰는 기능을 제공합니다. CSV 파일은 플랫 파일의 한 형태이므로, FlatFileItemWriter를 사용하여 CSV 파일을 생성할 수 있습니다.

 

FlatFileItemWriter는 Chunk 모델에서 청크 단위로 데이터를 받아 파일에 쓰는 역할을 담당합니다. write() 메서드는 List 형태의 아이템들을 받아 처리하며, 각 아이템은 LineAggregator를 통해 문자열로 변환됩니다. 변환된 문자열은 파일에 한 줄씩 기록됩니다.

 

Spring Batch가 Writer 계층에 CSV 생성 책임을 둔 이유는 관심사의 분리(separation of concerns)를 위해서입니다. 데이터 변환 로직(LineAggregator)과 파일 쓰기 로직(FlatFileItemWriter)을 분리함으로써, 각 컴포넌트의 재사용성과 테스트 가능성을 높입니다. 또한 다양한 파일 형식(CSV, 고정폭 파일 등)을 동일한 Writer 구조로 처리할 수 있도록 설계되었습니다.

LineAggregator

LineAggregator는 Spring Batch가 제공하는 인터페이스로, 아이템을 문자열로 변환하는 역할을 담당합니다. Spring Batch API Javadoc에 따르면, LineAggregator는 ItemWriter 내부에서 사용되어 각 아이템을 파일에 기록할 수 있는 형태로 변환합니다.

 

FlatFileItemWriter는 LineAggregator를 통해 각 아이템을 문자열로 변환합니다. CSV 파일의 경우 DelimitedLineAggregator를 사용하며, 이 클래스는 FieldExtractor를 사용하여 객체의 필드를 추출하고 구분자로 연결합니다. LineAggregator가 Writer 내부에서 수행하는 책임 범위는 데이터 변환에 한정됩니다. 파일 쓰기, 버퍼 관리, 리소스 관리 등의 책임은 FlatFileItemWriter가 담당합니다.

FlatFileHeaderCallback

FlatFileHeaderCallback은 Spring Batch가 제공하는 인터페이스로, 파일 헤더를 작성하는 역할을 담당합니다. Spring Batch API Javadoc에 따르면, 이 인터페이스는 FlatFileItemWriter에 설정되어 Step 실행 시 파일 헤더를 작성합니다.

 

FlatFileHeaderCallback의 writeHeader() 메서드는 Step 실행 흐름에서 파일 쓰기가 시작되기 전에 한 번 호출됩니다. 이 시점은 FlatFileItemWriter의 open() 메서드가 호출된 직후이며, 실제 데이터 쓰기가 시작되기 전입니다. CSV 파일의 경우 컬럼명을 헤더로 작성하는 데 사용됩니다.

 

    @Bean(name="OutboundInvoiceStep1Writer")
    @StepScope // .csv 데이터 생성
    public FlatFileItemWriter<CSVRow> OutboundInvoiceStep1Writer() {

        log.info("OutboundInvoiceStep1Writer started ... ");

        String releaseDate_HH = parameter.getReleaseDate().format(DefaultDateTimeFormat.DATE_NONE_DASH_FORMAT)
                .concat("_")
                .concat(parameter.getExecutionTime());

        // tracking_batch_yyyyMMdd_HH.csv (조회기준날짜_실행시각)
        String filePath = TRACKING_BATCH_CSV_PATH.concat(releaseDate_HH).concat(".csv");
        getInstance().setFilePath(filePath);

        return new FlatFileItemWriterBuilder<CSVRow>()
            .name("OutboundInvoiceStep1Writer")
            .resource(new FileSystemResource(filePath))
            .encoding("UTF-8")
            .delimited()
            .delimiter("\t")
            .names("orderNo", "invoiceNo")
            .lineAggregator(new OutboundInvoiceLineAggregator())
//            .headerCallback(new OutboundInvoiceHeader()) // .csv 헤더 제거
            .build();
    }

FlatFileItemWriter를 사용해 csv 데이터를 생성 

 

public class OutboundInvoiceLineAggregator implements LineAggregator<CSVRow> {
    @Override
    public String aggregate(CSVRow item) {
        return item.getOrderNo() + "," + item.getInvoiceNo();
    }
}

LineAggregator를 사용해 Item을 csv 로우로 변환 

 

public class OutboundInvoiceHeader implements FlatFileHeaderCallback {

    @Override
    public void writeHeader(Writer writer) throws IOException {
        writer.write("OrderNumber,TrackingNumber");
    }
}

FlatFileHeaderCallback을 사용해 csv 데이터의 헤더 추가 

 

[다이어그램3] csv item 쓰기

[다이어그램4] csv item 쓰기

 

 

 

---

 

API 응답결과를 CSV 데이터로 쓰는 배치 잡의 다이어그램

 

실무적 고려사항 및 팁

외부 API 기반 데이터 처리를 구현할 때는 다음과 같은 사항을 고려해야 합니다. 네트워크 오류나 API 응답 지연에 대한 예외 처리를 구현해야 합니다. AbstractPagingItemReader는 ItemStream 인터페이스를 구현하므로, 재시작 가능한 작업으로 구성하면 장애 발생 시 이전 상태에서 재개할 수 있습니다.

 

페이지 크기와 청크 크기를 조정하여 성능을 최적화할 수 있습니다. 페이지 크기가 너무 크면 메모리 사용량이 증가하고, 너무 작으면 API 호출 횟수가 증가하여 성능이 저하될 수 있습니다. 청크 크기도 마찬가지로 트랜잭션 오버헤드와 메모리 사용량 사이의 균형을 고려하여 설정해야 합니다.

 

CSV 파일 작성을 구현할 때는 데이터 일관성을 유지하기 위한 로직을 추가해야 합니다. FlatFileItemWriter는 청크 단위로 데이터를 쓰므로, 각 청크가 성공적으로 기록되었는지 확인하는 메커니즘을 고려할 수 있습니다. 또한 파일 쓰기 중 장애가 발생한 경우를 대비하여 임시 파일을 사용하거나, 쓰기 완료 후 원자적으로 파일을 이동하는 방식을 고려할 수 있습니다.

맺음말 및 요약

이 글에서는 Spring Batch의 커스텀 ItemReader와 ItemWriter 구현 방법을 설명했습니다. Spring Batch 공식 모델에 따르면, 배치 작업은 Chunk-oriented Processing을 통해 데이터를 읽고 처리하고 씁니다. AbstractPagingItemReader는 페이징 기반 데이터 읽기를 일반화하여, 외부 API와 같은 다양한 데이터 소스에 일관된 방식으로 접근할 수 있도록 설계되었습니다.

 

FlatFileItemWriter는 플랫 파일 쓰기를 담당하며, LineAggregator를 통해 데이터 변환 로직을 분리합니다. 이러한 설계는 관심사의 분리를 통해 각 컴포넌트의 재사용성과 테스트 가능성을 높입니다.

 

API 기반 배치 처리 시 핵심 체크 포인트는 다음과 같습니다. ItemStream 인터페이스를 구현하여 재시작 가능한 작업으로 구성해야 합니다. 페이지 크기와 청크 크기를 데이터 특성과 시스템 리소스를 고려하여 최적화해야 합니다. 네트워크 오류나 API 제한에 대한 예외 처리와 재시도 메커니즘을 구현해야 합니다. 이러한 고려사항을 반영하여 안정적이고 효율적인 배치 작업을 구성할 수 있습니다.