Develop/Spring+JPA

[Spring Batch] 개념부터 코드까지

연로그 2023. 7. 21. 23:11
반응형

목차

1. Spring Batch란?

2. Spring Batch 구조

3. 기본적인 세팅

4. Job, Step

5. ItemReader, ItemProcessor, ItemWriter

 

 

1. Spring Batch란?

배치란 언제, 왜 쓰는걸까? 비즈니스 로직을 작성하다 보면 대량 처리(bulk processing)가 필요한 경우가 많다. 이런 대량 처리가 반복적으로 일어나야 하는 경우, 배치 애플리케이션을 생성 및 실행하여 처리할 수 있다. 그 중에서도 Spring Batch는 엔터프라이즈 시스템에서 일상적인 운영에 필요한 배치 애플리케이션을 개발할 수 있도록 설계된 배치 프레임워크이다.

 

여기서 주의할 점은 배치는 대량의 데이터를 일괄적으로 처리할 뿐, 특정 주기마다 자동으로 돌아가는 스케줄링과는 관련이 없다. Spring Batch는 스케줄러와 함께 사용할 수 있도록 설계되어있을 뿐이지 스케줄러 자체를 대체하는 것은 아니다. (= 배치는 대량 처리를 의미하는 것이지, 일정 주기마다 돌아가는 것을 의미하지 않는다.) 따라서 작업 스케줄링 라이브러리인 Quartz 등과 Spring Batch를 비교하는 것은 적절하지 못하다.

 

batch의 뜻을 잊지 말자.

 

 

2. Spring Batch Architecture

 

Spring Batch의 Architecture를 간단히 살펴보자. 여기서 Job, Reader, Writer 등의 생소한 용어들이 나오는데 요 부분은 4, 5에서 다루겠다. 이런 구성 요소가 있나보다~ 정도로 넘어가자.

 

Spring Docs - Spring Batch Layered Architecture

 

  • Application: 개발자가 작성한 모든 배치 Job과 코드들
  • Core: 배치 작업을 시작하고 제어할 때 필요한 핵심 Runtime Classes (ex: JobLauncher, Job, Step)
  • Infrastructure: 애플리케이션 개발자와 코어 프레임워크가 사용하는 공통 Reader, Writer, Service를 포함

 

 

3. 기본적인 세팅

 

본격적인 Spring Batch 설명에 들어가기 앞서, 예제 코드를 작성하기 위해 준비해야 하는 것들이 있다. 먼저 Spring Batch 의존성을 추가해두어야 한다. Gradle의 경우 dependencies에 아래 코드를 추가하면 된다.

implementation 'org.springframework.boot:spring-boot-starter-batch'

 

그리고 메인 메서드가 포함된 Application에 아래와 같이 @EnableBatchProcessing을 추가하자.

@EnableBatchProcessing
@SpringBootApplication
public class BatchApplication {
    public static void main(String[] args) {
        SpringApplication.run(BatchApplication.class, args));
    }
}

 

 

4. Job, Step

 

Job

  • 배치 처리 과정을 하나의 단위로 만든 객체
  • 배치 처리 과정에 있어 전체 계층의 최상단에 위치
  • 1개 이상의 Step을 가짐

 

📝 예제코드 1 - 기본 예제

@Bean
public Job sendNotificationJob(JobRepository jobRepository) {
    return new JobBuilder("sendNotificationJob", jobRepository)
                    .start(sendMailStep())  // mail을 전송하는 step 등록
                    .next(sendSnsStep())    // sns를 전송하는 step 등록
                    .build();
}
  • JobRepository
    • Spring Batch 내에서 사용되는 다양한 도메인 객체들의 기본적인 CRUD 작업에 이용
    • Job을 생성하려면 기본적으로 필요.
    • Spring Batch 세팅 시 추가했던 @EnableBatchProcessing을 사용하면 자동으로 제공
  • start(), next()
    • Step을 추가하기 위한 메서드
    • 이전 Step의 작업이 완료되어야만 다음 Step이 실행됨

 

📝 예제코드 2 - JobBuilderFactory 이용하기 (deprecated)

@Bean
public Job sendNotificationJob(JobBuilderFactory jobBuilderFactory) {
    return jobBuilderFactory.get("sendNotificationJob")
                    .start(sendMailStep())  // mail을 전송하는 step 등록
                    .next(sendSnsStep())    // sns를 전송하는 step 등록
                    .build();
}
  • JobBuilderFactory
    • Spring Batch 5.0.0에서 deprecated
    • JobRepository를 직접 넘기지 않고도 Job 생성
    • get 메서드 내부를 살펴보면, JobRepository를 넣어주고 있음

JobBuilderFactory

 

Step

  • 배치 작업을 정의 및 제어하고, 순차적인 단계를 캡슐화한 것
  • 개발자가 설계한 것에 따라 복잡해질 수도, 간단해질 수도 있음
  • ItemReader, ItemProcessor, ItemWriter로 구성

Spring Docs - Step

 

📝 예제코드 1 - 기본 예제

@Bean
public Step sendMailStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
    return new StepBuilder("sendMailStep", jobRepository)
                .<MemberDto, Long>chunk(10, transactionManager)
                .reader(itemReader())
                .process(itemProcessor()) // optional. 필수가 아니다.
                .writer(itemWriter())
                .build();
}
  • JobRepository: Job과 동일.
  • PlatformTransactionManager: 처리 중의 트랜잭션 시작/커밋을 담당하는 Spring의 트랜잭션 매니저
  • chunk: 하나의 트랜잭션이 처리할 item의 개수

 

📝 예제코드 2 - StepBuilderFactory 이용하기 (deprecated)

@Bean
public Step sendMailStep(StepBuilderFactory stepBuilderFactory) {
    return stepBuilderFactory.get("sendMailStep")
                .<MemberDto, Long>chunk(10)
                .reader(itemReader())
                .process(itemProcessor()) // optional. 필수가 아니다.
                .writer(itemWriter())
                .build();
}
  • StepBuilderFactory
    • Spring Batch 5.0.0에서 deprecated
    • JobRepository와 PlatformTransactionManager를 get() 안에서 세팅해주고 있음

StepBuilderFactory

 

5. ItemReader, ItemProcessor, ItemWriter

 

Spring Batch는 Job을 실행시키고, Job은 1개 이상의 Step으로 구성된다. 그리고 Step은 ItemReader, ItemProcessor, ItemWriter로 구성된다는 것을 확인했다. 각 요소에 대해 간단한 예제 코드를 살펴보자.

 

📝 예제 코드 - ItemReader

private DataSource dataSource; // bean을 주입 받았다는 가정 하에 사용

// DB에서 회원 정보를 조회하는 ItemReader
@Bean
public ItemReader<MemberDto> itemReader() throws Exception {
    return new JdbcPagingItemReaderBuilder<MemberDto>()
            .pageSize(10)
            .dataSource(dataSource)
            .rowMapper(new BeanPropertyRowMapper<>(MemberDto.class)) // 값 매핑이 제대로 되지 않으면 getter 호출 시 빈값("")이 반환될 수 있음
            .queryProvider(getQueryProvider())
            .name("itemReader")
            .build();
}

private PagingQueryProvider getQueryProvider() throws Exception {
        var queryProvider = new SqlPagingQueryProviderFactoryBean();
        queryProvider.setDataSource(dataSource);

        queryProvider.setSelectClause("id, name, email"); // 조회할 컬럼
        queryProvider.setFromClause("member");            // 테이블명
        
        // SqlPagingQueryProviderFactoryBean의 경우, 페이징 처리로 인해 sortKey 설정이 필수적.
        queryProvider.setSortKey("id");  

        return queryProvider.getObject();
}
  • ItemReader: 입력 데이터, 파일, DB 등등 다양한 소스로부터 item을 읽는 역할
  • JdbcPagingItemReader 외에도 HibernatePagingItemReader, JpaPagingItemReader 등 다양한 ItemReader 지원 (물론 직접 구현하는 것도 가능)

 

ItemReader를 직접 구현하는 경우, 주의할 점이 있다. ItemReader에서 item을 가져올 때 read() 메서드를 호출하는데, 이때 null 값이 반환되면 더이상 호출할 값이 없다고 판단한다. ItemProcessor 또는 ItemWriter 과정에서 값을 읽다말고 끝나버린다면, ItemReader에서 null 값이 반환되는 케이스가 있을지 의심해보아야 한다.

Returns: T the item to be processed or null if the data source is exhausted     ㅡ ItemReader::read 메서드 설명의 일부

 

 

📝 예제 코드 - ItemProcessor

@Bean
public ItemProcessor<MemberDto, Long> itemProcessor() {
    return member -> Long.parseLong(member.getId());
}

 

ItemProcessor는 필수가 아닌 선택이다. 개인적으로는 매핑 작업이나 데이터 필터링이 필요할 때 사용하고 있다. 예를 들어 ItemReader에서 MemberDto 형태로 데이터를 조회했는데, ItemWriter에서는 MemberDto 내부의 Long id 값만 필요하다면? ItemProcessor에서 MemberDto에서 id 값을 꺼내 Long으로 변환하는 작업을 거쳤다. 또 다른 예로는 ItemReader에서 전체 회원 목록을 조회했는데, 이 중에서도 20살 이상의 회원들에게만 작업이 진행되어야 한다면? ItemProcessor에서 age >= 20인 사람을 필터링하는 작업을 추가해주었다.

 

 

📝 예제 코드 - ItemWriter

@Bean
public ItemWriter<Long> itemWriter() {
    return memberIds -> memberIds.forEach(it -> sendMail(it));
}

private void sendMail(Long id) {
    // 메일 전송하는 로직
}

 

이번에는 Step이 어떤 플로우로 실행되는지 살펴보면서 각각의 역할을 알아보자.

 

Spring Docs - Chunk-oriented Processing

 

Step이 실행된다면 위 이미지와 같은 흐름이 진행된다.

  1. read()를 통해 item을 읽어온다.
  2. (ItemProcessor가 존재한다면) process()을 통해 item을 처리한다.
  3. write()를 통해 items을 사용한다.

 

위 그림을 다시한번 살펴보자. 의아한 점이 있지 않은가? read(), process()는 여러번 실행됐는데 write()는 한번만 실행됐다. 자세히 살펴보면 read()에서는 item을 읽고, process() 호출할 때도 item을 넘겼는데 write()에서는 item's'를 넘겼다. 요 부분에 대해서는 Spring Batch Docs에서 작성해준 의사 코드를 보면 이해하기 쉽다.

// 코드 출처: https://docs.spring.io/spring-batch/docs/current/reference/html/step.html#configureStep

List items = new Arraylist();

// commitInterval 수만큼 item을 읽어온다.
for(int i = 0; i < commitInterval; i++){
    Object item = itemReader.read();
    if (item != null) {
        items.add(item);
    }
}

// items를 하나씩 순회하며 ItemProcessor를 처리한다.
List processedItems = new Arraylist();
for(Object item: items){
    Object processedItem = itemProcessor.process(item);
    if (processedItem != null) {
        processedItems.add(processedItem);
    }
}

// item 목록을 ItemWriter에 넘겨 ItemWriter의 로직을 수행한다.
itemWriter.write(processedItems);

 

 

이제 Spring Batch를 작성하기 위한 최소한의 지식은 정리를 마쳤다. Spring Batch에 대해서 더 공부하고 싶다면 JobExecution, flow, Chunk, RetryTemplate 등의 키워드를 참고해보자. 특히 Spring Batch는 chunk-oriented processing이므로Chunk에 대한 이해도가 높으면 좋을 것 같다. 이후 chunk에 대한 포스팅을 작성하면 링크를 추가하겠습니다. 🙂

참고

반응형