Spring Batch 5 - 5. ItemReader

이전 게시글

https://wn1331.tistory.com/255

Spring Batch 5 - 4. Task 기반 step, Chunk 기반 step

 

Chunk기반 Step에서 ItemReader에 대한 포스팅이다. 

 

1. ItemReader란?

ItemReader는 하나의 인터페이스이고, 아래와 같이 정의되어 있다.

이를 직접 구현하여 ItemReader를 직접 만들 수 있다. 우아한형제들에서는 JpaPagingItemReader를 본따 QuerydslZeroOffsetItemReader를 ItemReader 구현체로 직접 구현하여 사용하고 있다. (아마 스프링배치5 마지막 포스팅은 저걸 따라서 구현해 보는게 되지 않을까 싶다..)

 

스프링에서 제공하는 ItemReader 인터페이스의 구현체는 여러 가지가 있지만, File, XML, JSON, JMS 등과 같은 데이터들을 다루는 Reader들은 다루지 않겠다.

 

2. Paging기반과 Cursor기반

스프링 프레임워크의 강점 중 하나는 개발자가 비즈니스 로직에만 집중할 수 있도록 JDBC와 같은 문제점을 추상화한 것이다.

이를 통해 스프링 배치 개발자들은 스프링 프레임워크의 JDBC 기능을 확장하였다.

 

일반적으로 대용량 데이터나 대규모 데이터를 처리해야 할 때 배치 어플리케이션이 사용된다.

 

수백만 개의 데이터를 조회하는 쿼리가 있는 경우, 모든 데이터를 한 번에 메모리에 불러오는 것은 효율적이지 않다. Spring의 JdbcTemplate은 분할 처리를 지원하지 않기 때문에 개발자가 직접 limit, offset 등을 사용하여 작업해야 한다.

스프링 배치에서는 이런 문제를 해결하기 위해 2개의 Reader 타입을 제공한다.

 

1. Cursor:

Cursor는 실제로 JDBC ResultSet의 기본 기능이다.

ResultSet이 열릴 때마다 next() 메서드가 호출되어 데이터베이스에서 데이터가 반환된다. 이를 통해 필요에 따라 데이터베이스에서 스트리밍 방식으로 데이터를 읽어올 수 있다. 일반적으로 Thread-Safe하지 않다.

2. Paging:

Paging은 좀 더 많은 작업을 필요로 한다. 

페이지 단위로 한 번에 일정량의 데이터(Chunk 혹은 Pagesize)를 조회하는 방식인데, 각 페이지는 limit과 offset 값을 사용하여 데이터베이스에서 검색된다. 일반적으로 Thread-Safe하며, 코드레벨에서 매우 안정적이지만 좀더 많은 작업을 요구한다. ChunkSize와 PageSize는 같게 해주어야 머리가 덜 아플 것이다.

 

Cursor와 Paging 중 어떤 Reader 타입을 선택할지는 상황과 요구사항에 따라 결정된다.

Cursor는 Streaming 방식으로 동작해서 대용량 또는 실시간 처리가 필요한 경우 유용하며, Paging은 Chunk 단위로 일괄 처리해야 할 때 유용하고, 한 번에 메모리에 로드되지 않고 페이징 처리되므로 대량의 데이터 처리 시 성능상 이점을 가진다.

 

각 Reader 타입은 스프링 배치에서 제공하는 구현체인 JdbcCursorItemReader와 JdbcPagingItemReader 등으로 사용할 수 있고, 필요한 경우 커스텀 ItemReader 클래스도 개발할 수 있다.

 

Cursor와 Paging을 그림으로 비교하면 다음과 같다.

Paging 방식에서 10row는 PageSize를 뜻한다. 

Cursor 방식은 Database와 커넥션을 맺은 후, Cursor를 한 칸씩 옮기면서 지속적으로 데이터를 받아온다.

반면 Paging 방식에서는 한번에 개발자가 지정한 PageSize 만큼 데이터를 가져온다.

 

정리 : 

1. Cursor방식 - Thread-Safe하지 않음, Streaming방식. 한줄씩 읽음(ResultSet)

1. Paging방식 - Thread-Safe, Paging방식, 많은작업 요구. 여러 줄씩 읽음(PageSize만큼)

Cursor 방식도 SynchronizedItemStreamReader로 한번 감싸면 Thread-Safe하도록 동기화처리를 할 수 있다. 

아래는 그 예시이다.

@Bean(name = MAILING_JOB_NAME)
@StepScope
public SynchronizedItemStreamReader<Product> reader(){

return new SynchronizedItemStreamReaderBuilder<Product>() 
        .delegate(사용할 Thread-Not-Safe한 ItemReader 구현체) // (1)
        .build();
}

 

2가지 방식의 구현체는 다음과 같다.

• Cursor 기반의 ItemReader 구현체

• JdbcCursorItemReader
• JpaCursorItemReader

• Paging 기반의 ItemReader 구현체

• JdbcPagingItemReader
• JpaPagingItemReader                                                                                             

 

1. JdbcCursorItemReader

JdbcCursorItemReader는 ResultSet과 함께 사용되고, Datasource로부터 SQL을 통해 값을 불러온다. 쓰레드 안정성을 보장하지 않기 때문에 멀티쓰레드 환경에서 사용한다면, 동기화 처리를 통한 동시성 문제를 해결해주어야 한다. (SynchronizedItemStreamReader)

 

public JdbcCursorItemReader<Employee> jdbcCursorReader() {
    return new JdbcCursorItemReaderBuilder<Employee>()
            .name("jdbcCursorReader")
            .fetchSize(20)
            .dataSource(dataSource)
            .beanRowMapper(Employee.class)
            .sql("select empno, ename, job, hiredate, sal, deptno from emp")
            .maxItemCount(20)
            .currentItemCount(0)
            .build();
}

 

2. JpaCursorItemReader

 

public JpaCursorItemReader<Employee> jpaCursorReader() {
    return new JpaCursorItemReaderBuilder<Employee>()
            .name("jpaCursorItemReader")
            .entityManagerFactory(emf)
            .queryString("select e.empNo, e.eName, e.job, e.hireDate, e.sal, e.deptNo from Employee e")
            .currentItemCount(0) //조회 item의 시작지점
            .maxItemCount(100) //조회할 최대 item 수
            .build();
}

 

3. JdbcPagingItemReader

public JdbcPagingItemReader<Employee> jdbcPagingReader() throws Exception {
    JdbcPagingItemReader<Employee> jdbcPagingReader = new JdbcPagingItemReaderBuilder<Employee>()
            .name("jdbcPagingReader")
            .pageSize(3) //페이지의 사이즈를 설정해준다. (쿼리 당 요청할 레코드 수)
            .fetchSize(20)
            .dataSource(dataSource)
            .beanRowMapper(Employee.class) //객체 클래스를 넣으면 자동으로 DB 데이터가 객체에 맵핑됨.
            .maxItemCount(10)//한번에 조회할 최대 item 수 설정
            .currentItemCount(0)//조회 Item의 시작 지점 설정
            .queryProvider(createQueryProvider())
            .build();

    jdbcPagingReader.afterPropertiesSet();
    return jdbcPagingReader;
}

 

4. JpaPagingItemReader

public JpaPagingItemReader<Employee> jpaPagingReader() {
    return new JpaPagingItemReaderBuilder<Employee>()
            .name("jpaPagingReader")
            .entityManagerFactory(emf)
            .queryString("select e.empNo, e.eName, e.job, e.hireDate, e.sal, e.deptNo from Employee e")
            .pageSize(5)
            .build();
}