❐ 1. ItemReader

itemReader란?
  • ItemReader는 여러 종류의 입력 데이터 소스로부터 데이터를 읽어오는 역할을 하는 인터페이스
  • 대표적인 구현 예시
    • Flat File
      • CSV, TSV 같은 평문 파일에서 한 줄씩 데이터를 읽어옵니다.
      • 각 라인은 하나의 레코드이며, 필드는 쉼표(,) 등 구분자로 나뉩니다.
    • XML
      • XML 파일에서 데이터를 읽습니다.
      • XML의 각 요소를 파싱하고 매핑하거나, XSD 스키마로 유효성 검사를 수행할 수도 있습니다.
    • Database
      • 데이터베이스에서 ResultSet을 읽어와 객체로 매핑합니다.

 

 

인터페이스 정의
public interface ItemReader<T> {
    T read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException;
}
  • read() 메서드는 한 번 호출될 때마다 하나의 아이템을 반환하거나, 더 이상 읽을 게 없으면 null을 반환
  • 반환되는 데이터는 파일의 한 줄, DB의 한 행, XML의 한 요소 등일 수 있습니다.
  • 이 아이템은 보통 도메인 객체(Trade, Foo 등)로 매핑되지만, 반드시 그럴 필요는 없습니다.

 

 

동작 방식
  • ItemReader는 forward-only(단방향) 으로 동작합니다.
    • 즉, 읽은 데이터를 다시 되돌리거나 이전 데이터를 재참조할 수 없습니다.
  • 하지만, 트랜잭션 자원(JMS queue 등)을 사용하는 경우, 롤백 시 동일한 아이템을 다시 읽을 수 있습니다.
    • 트랜잭션 롤백 시 이전 상태로 돌아가므로, 동일한 데이터가 재처리되는 것이죠.
  • 예외 처리
    • ItemReader가 단순히 읽을 데이터가 없어서 결과가 0건인 경우엔 예외를 던지지 않습니다.
    • 예를 들어 DB Reader가 쿼리 결과가 없을 경우, 첫 번째 read() 호출에서 단순히 null을 반환합니다.

 

 

 

 

 

❐ 2. ItemWriter

ItemWriter란?
  • ItemWriter는 ItemReader와 기능적으로 유사하지만, 역방향의 역할을 수행합니다.
    • 즉, 데이터를 읽어오는 대신 데이터를 외부로 내보내는(write) 역할을 합니다.
  • Reader와 마찬가지로 Writer도 리소스(DB, 파일, 큐 등) 를 찾고, 열고, 닫는 과정이 필요합니다.
  • 하지만 차이점은,
    • Reader는 데이터를 읽어들이는(read) 반면
    • Writer는 데이터를 기록하는(write) 역할을 합니다.
      • 예를 들어, DB나 메시지 큐를 사용하는 경우에는 insert, update, send 등의 작업

 

인터페이스 정의
public interface ItemWriter<T> {
    void write(Chunk<? extends T> items) throws Exception;
}
  • write() 메서드는 하나 이상의 아이템(Chunk 단위) 을 받아서 외부로 기록합니다.
  • Exception을 던질 수 있으며, 내부적으로는 파일에 쓰거나 DB에 insert/update 등을 수행할 수 있습니다.

 

동작 방식
  • ItemReader의 read() 메서드처럼, ItemWriter의 핵심 계약(contract)은 write()입니다.
  • 이 메서드는 열려 있는 동안(open) 아이템 리스트를 출력하려고 시도합니다.
  • 일반적으로 아이템들은 하나씩이 아니라 Chunk 단위로 모여서 전달됩니다.
    • 즉, write(List<T> items) 가 아니라, “한 묶음(Chunk)”으로 처리됩니다.
    • 이는 성능을 위해 데이터를 모아서 한번에 처리하기 위함입니다.

 

flush 처리
  • 데이터를 모두 쓴 뒤에는, 필요하다면 flush 작업을 수행할 수 있습니다.
    • (예: 버퍼에 남은 데이터나 세션 캐시를 실제 DB에 반영)
  • 예시로 Hibernate 기반 Writer의 경우
    • 여러 번의 write() 호출 후,
    • Hibernate Session의 flush()를 호출하여 실제 DB 반영을 수행할 수 있습니다.
// JpaItemWriter.java
@Override
public void write(Chunk<? extends T> items) {
    EntityManager entityManager = EntityManagerFactoryUtils.getTransactionalEntityManager(entityManagerFactory);
    if (entityManager == null) {
        throw new DataAccessResourceFailureException("Unable to obtain a transactional EntityManager");
    }
    doWrite(entityManager, items);
    entityManager.flush();
    if (this.clearPersistenceContext) {
        entityManager.clear();
    }
}

 

 

 

 

 

❐ 3. ItemStream

itemStream이란?
  • ItemReader와 ItemWriter의 공통적인 “상태 관리(State Management)” 문제를 해결하기 위한 인터페이스
  • Batch Job에서는 Step이 실패하거나 재시작될 수 있기 때문에
    • 현재까지 어디까지 읽었는지, 어떤 리소스를 열었는지를 기억해야 함.

 

인터페이스 정의
public interface ItemStream {
    void open(ExecutionContext executionContext) throws ItemStreamException;
    void update(ExecutionContext executionContext) throws ItemStreamException;
    void close() throws ItemStreamException;
}
  1. open
    • Step이 시작될 때 호출
    • 파일 핸들 열기, DB 커넥션 생성 등 리소스 초기화 작업을 수행
    • ItemReader의 경우 read()를 호출하기 전에 반드시 open()이 호출되어야 함.
  2. update
    • Step 실행 중 주기적으로 호출되어, 현재 상태를 ExecutionContext에 저장
    • 즉, “현재 몇 번째 아이템까지 읽었는가?”, “현재 커서 위치는 어디인가?” 같은 상태 정보를 저장
    • 보통 트랜잭션 커밋 직전에 호출되어, 다음 재시작 시 복원할 수 있도록 함.
  3. close
    • Step이 끝나면 호출되어, open() 때 열었던 리소스를 닫음.
    • 파일 스트림, DB 커넥션 등을 안전하게 정리

 

 

 

 

 

❐ 4. The Delegate Pattern and Registering with the Step

한마디로, 위임 객체는 직접 주입해서 사용해야 한다.
  • CompositeItemWriter는 위임(Delegation) 패턴의 예로, Spring Batch에서 흔히 사용됨.
  • 위임 객체(delegate) 자체는 StepListener 같은 콜백 인터페이스를 구현할 수 있음.
  • 위임 객체들은 Step이 직접 알지 못하므로, listener나 stream으로 주입해서 등록해야 함.

 

CompositeItemWriter란?
  • "하나의 Writer 안에서 여러 Writer를 순차적으로 실행" 하기 위한 기능
  • 보통은 하나의 Step에서는 하나의 writer를 쓰는데, 만약 여러개의 writer를 써야한다면?
    • 예상 시나리오 : DB에 저장하면서 동시에 로그 파일도 남겨야 함.
//Composite: 두 writer를 순서대로 실행 (DB → 파일)
@Bean
fun compositeWriter(): CompositeItemWriter<UserDto> =
    CompositeItemWriter<UserDto>().apply {
        setDelegates(listOf(jdbcWriter(), fileWriter()))
    }

 

 

 

 

 

❐ 5. Database

🌀5-1. Cursor-based ItemReader Implementations

Spring JDBC의 동작 방식과 Spring Batch의 근본적인 설계 차이
구분 JdbcTemplate JdbcCursorItemReader(Batch)
데이터 읽기 방식 한 번에 전부 읽음 한 행씩 순차적으로 읽음
ResultSet 유지 시간 메서드 내부에서만 Step 전체 동안 유지
닫히는 시점 메서드 종료 시 Step 완료 시
장점 빠르고 안전한 리소스 해제 대용량 처리에 유리 (스트리밍)
단점 대용량일 경우 메모리 부담 Connection을 오래 유지해야 함
  • Spring의 JdbcTemplate은 콜백(RowMapper)으로
    • 모든 행을 한 번에 매핑하고 ResultSet과 Connection을 메서드 안에서 바로 닫음.
    • 따라서, 리소스 누수나 장시간 Connection 점유 문제가 없음.
  • 하지만 Spring Batch에서는 Step 전체 동안 ResultSet을 열어둬야 함.
    • 따라서, 닫는 시점을 “Step 완료 시점”까지 지연시켜야 함.

 

Spring Batch의 JdbcCursorItemReader가
어떻게 ResultSet 커서를 움직이며 데이터를 읽는지 를 설명하는 예시

  • 커서(cursor) 는 ID=2에서 시작해서 ID=6까지 한 행씩 순차적으로 이동
  • 각 행(Foo 객체)은 한 번 쓰기(write())가 끝나면 더 이상 참조되지 않음 ➔ GC가 메모리에서 정리할 수 있음.
    • 대용량 데이터를 처리해도 메모리 부담이 거의 없음.

 

JdbcCursorItemReader
    • 배치 작업에서 대규모 데이터를 한 번에 메모리로 가져오면 메모리 사용량과 트랜잭션 범위가 급격히 커짐
    • JdbcCursorItemReader는 
      • JdbcCursorItemReader는 커서 기반 Reader의 JDBC 구현
      • SQL을 직접 실행해 ResultSet을 순방향으로 스트리밍
      • 데이터베이스 커서 기반으로 행을 한 건씩 스트리밍해 읽고,
              Spring Batch가 청크 단위로 "읽기/쓰기/커밋"을 관리하도록 해줌.
  • 고성능 배치 처리에 적절함.

 

JdbcTemplate과의 차이
  • JdbcTemplate은 보통 조건에 맞는 모든 행을 조회하여 객체 리스트로 메모리에 적재
    • 행이 1,000개면 객체도 1,000개가 메모리에 생성
  • JdbcCursorItemReader는 read()로 한 건을 가져와 쓰고, 다시 read()로 다음 건을 가져오는 스트리밍 흐름
  • 결과물은 같아도 처리 방식이 다름.
    • JdbcTemplate : 일괄 적재
    • JdbcCursorItemReader : 스트리밍

 

StoredProcedureItemReader
  • 저장 프로시저(또는 함수)를 호출해 커서 데이터를 조회함.

 

 

🌀5-2. Paging ItemReader Implementations

페이징 방식으로 데이터 가져오기
  • 데이터베이스 커서를 사용하는 대신, 결과를 여러 개의 쿼리로 나누어 가져오는 방법
  • 이때 각 쿼리가 가져오는 결과의 일부를 ‘페이지(page)’라고 부름.
  • DB Connection을 오래 잡지 않음 (매 쿼리마다 닫힘)
  • 페이징으로 나눠서 처리하므로 안정적
  • 각 쿼리는 시작 행 번호와 해당 페이지에서 반환받을 행의 개수를 반드시 지정해야 함.

 

JdbcPagingItemReader
  • JdbcPagingItemReader는 ItemReader의 구현체 중 하나
  • 쿼리를 여러 번 실행해 “페이지 단위”로 데이터를 읽는다.
  • 페이지를 구성하는 행을 조회하는 SQL을 제공하는 PagingQueryProvider가 필요
    • 지원하는 각 데이터베이스 유형에 대해 서로 다른 queryProvider를 사용해야함.
@Bean
fun queryProvider(): SqlPagingQueryProviderFactoryBean {
    val provider = SqlPagingQueryProviderFactoryBean()
    provider.setSelectClause("select id, name, credit")
    provider.setFromClause("from customer")
    provider.setWhereClause("where status = :status")
    provider.setSortKey("id")
    return provider
}
  • 내부적으로 쿼리 생성에 SqlPagingQueryProviderFactoryBean을 사용
  • sortKey는 유일(unique) 해야 함. 그래야 실행 중 중복 데이터나 누락이 발생하지 않음

 

 

JpaPagingItemReader
  • JPA가 자체적으로 페이징(setFirstResult, setMaxResults) 기능을 제공
  • 따라서 Spring Batch는 이 기능을 그대로 활용해서 데이터를 읽는다.
  • 각 페이지를 읽은 뒤 엔티티를 Detach 시켜 JPA 캐시에서 제거
    • Hibernate에는 StatelessSession이라는 기능이 있어서
            1차 캐시(Session)를 사용하지 않고 엔티티를 읽을 수 있음
    • 하지만 JPA는 이런 기능이 없기 때문에
            페이징과 영속성 컨텍스트 초기화(clear) 를 통해 메모리 누수를 방지
    • 이렇게 하면 다음 페이지로 넘어갈 때,
            이전 엔티티들이 GC(Garbage Collection)될 수 있어 메모리 부담이 줄어듬.
@Bean
fun jpaPagingItemReader(emf: EntityManagerFactory): JpaPagingItemReader<Customer> =
    JpaPagingItemReaderBuilder<Customer>()
        .name("customerReader")
        .entityManagerFactory(emf)
        .queryString("SELECT c FROM Customer c WHERE c.status = :status ORDER BY c.id")
        .parameterValues(mapOf("status" to "ACTIVE"))
        .pageSize(500)
        .build()
더보기

🤔 JpaPagingItemReader

// JpaPagingItemReader.java
// 일부 내용 생략
public class JpaPagingItemReader<T> extends AbstractPagingItemReader<T> {

	private EntityManager entityManager;

	private String queryString;

	private JpaQueryProvider queryProvider;

	private Query createQuery() {
        if (queryProvider == null) {
            return entityManager.createQuery(queryString); // JPQL만 인식함
        }
        else {
            return queryProvider.createQuery();
        }
    }
}
  • 소스코드를 보면 queryProvider가 분기의 기준이 됨.
  • 즉, queryProvider가 정의되어 있지 않다면 entityManager가 JPQL을 바탕으로 쿼리를 생성함

 

그렇다면 native 쿼리를 사용할 순 없을까?
@Bean
fun jpaNativeProvider(): JpaNativeQueryProvider<SomeEntity> =
    JpaNativeQueryProvider<StakingProductEntity>().apply {
        setSqlQuery("SELECT tn.* FROM table_name as tn")
        setEntityClass(SomeEntity::class.java)
    }
    
@Bean(name = ["${JOB_NAME}Reader"])
fun reader(): JpaPagingItemReader<SomeEntity> {
    return JpaPagingItemReaderBuilder<SomeEntity>()
        .name("${JOB_NAME}Reader")
        .entityManagerFactory(entityManagerFactory)
        .queryProvider(jpaNativeProvider())
        .pageSize(CHUNK_SIZE)
        .build()
}
  • 위와 같이 커스텀 QueryProvider를 구현 후, queryProvider에 주입 시켜줘야 함.

 

 

🌀5-3. Database ItemWriters

  • 데이터베이스에는 파일 기반의 ItemWriter와 1:1로 대응되는 개념이 없음.
  • 파일은 자체적으로 트랜잭션을 보장하지 않기 때문에 별도의 ItemWriter 구현이 필요
    • 반면에,데이터베이스는 트랜잭션 매니저를 통해 커밋/롤백을 관리
  • 배치 처리에서 쓰기(write) 과정에서 오류가 발생하면 트랜잭션 경계 내에서 롤백이 이뤄져야 하며,
    • 플러시(flush) 타이밍이 에러 처리의 핵심 포인트

 

오류가 flush 시점에 발생한 경우

  • item들이 flush되기 전에 buffer에 저장된 경우, 이 시점에는 어떠한 error도 발생하지 않음.
  • flush가 호출되면 buffer가 비워지고, 이 시점에 예외가 발생함 (위 이미지 예시에서)
  • 이 때 Step은 할 수 있는게 없고, 트랜잭션이 롤백되어야 함.
  • 일반적으로는 장애가 난 지점을 재시도/스킵하는 정책을 쓸 수 있음.
  • 하지만 batch에서는 어떤 지점(item)에서 장애가 발생한지 알 수 없음
    • 왜? 버퍼에 있던 모든 item이 한번에 flush 됐기 때문 

 

 

오류가 write 시점에 발생한 경우

  • flush를 하는 경우 어느 지점에서 장애가 발생하는지 모르는데, 이걸 해결하기 위한 유일한 방법
  • 바로 각 item마다 flush를 호출하는 것!
  • (특히 Hibernate를 사용하는 경우) 가장 일반적인 usecase 
  • 이렇게 하면 장애가 나도, 어디서 났는지 알 수 있고 사후 처리(재시도 or 스킵)를 할 수 있음.

'Book > Spring Batch docs' 카테고리의 다른 글

Scaling and Parallel Processing  (0) 2025.10.10
Item processing  (0) 2025.10.09
Configuring a Step  (0) 2025.10.06
Configuring and Running a Job  (0) 2025.07.17
The Domain Language of Batch  (0) 2025.07.07

티스토리툴바