🌱Spring Batch (2) - Tasklet 방식, Chunk 방식, 그리고 간단한 구현

Spring Boot Batch 종류가 무엇이 있을까?🧐

Step이 구성하는 대표적인 두가지 방식이 존재한다.

• Tasklet 방식
• Chunk 방식

---

Tasklet 방식

Tasklet 방식은 Tasklet을 이용한 Task 기반의 처리 방식이며,
Batch의 Step 단계에서 단일한 레코드(row)나 파일을 하나의 작업만 처리하는 방식을 의미한다. Tasklet 방식은 각각의 처리를 하나의 트랜잭션으로 처리한다.

• 일반적인 수행 절차가 파일을 읽고 처리한 다음 결과를 데이터베이스에 쓰는 작업을 수행하며 이러한 단일 작업을 처리하기 때문에 작업이 끝날 때까지 대기해야 한다. 즉, 대용량 데이터 처리에는 적합하지 않다.

Chunk 방식

Chunk 방식은 Chuck을 이용한 Chunk 기반 처리 방식이며, Batch의 Step 단계에서 단일 한 레코드를 묶어서 여러 작업을 처리하는 방식을 의미한다.
Chunk방식은 묶인 레코드를 하나의 트랜잭션으로 처리하며 실패를 하는 경우 롤백을 수행한다.

• Chunk 방식은 병렬 처리를 위해 Chunk를 사용하되 순차적으로 처리하는 방식이다.
• Parallel Chunk 방식은 Chunk를 독립적으로 처리하여 병렬적인 처리를 수행한다.
• 대용량 데이터를 처리할 때 성능이 향상되고, 중복 레코드 처리나 실패한 레코드 처리 등 예외 상황에 대한 대처가 용이하다. <=>Chunk 방식과의 차이점
• Ex. 파일을 읽어 들여 데이터를 처리하는 작업이나 DB에서 데이터를 조회하여 처리하는 작업 등이 존재한다.

Chunk란?

Chunk란 데이터를 일정한 크기로 나눈 데이터 셋(Data Set)을 의미한다. 즉, 각 커밋 사이에 처리될 row(item) 수를 의미한다.

• Chunk 단위로 나누게 되면 전체 데이터를 한번에 처리하지 않기 때문에 메모리 부하를 줄이고 성능을 향상시킬 수 있다.
• 성공 시 Chunk만큼 커밋, 실패 시 Chunk만큼 롤백을 수행한다.

'단일한 레코드를 묶어서 처리한다'의 의미

• Chunk 방식에서는 ‘Reader-Processor-Writer 방식’을 이용한다.
• 이러한 처리 방식은 대용량 처리를 위해서 사용한다.

• ‘Reader’는 데이터를 읽어 들이는 역할을 한다.
• ‘Processor’는 읽어 들인 데이터를 가공하거나 필터링하는 역할을 한다.
• ‘Writer’는 가공된 데이터를 저장하는 역할을 수행한다.

Chunk 기반의 Step의 Reader-Processor-Writer 처리 방식의 흐름

1. 데이터 베이스에서 데이터를 읽어온다. (Item Reader)
2. 읽어온 데이터를 처리한다. (Item Processor)
3. 데이터를 저장한다. (Item Writer)

ItemReader

Cursor와 Paging 기능을 제공한다.

ItemProcessor

• Chunk-oriented Tasklet을 구성할 떄 선택 요소이다.
• 데이터를 가공/필터링 하는 역할을 수행(writer에서도 구현 가능한 역할을 하며 비즈니스 코드가 섞이는 것을 방지한다.)하며 Step에 여러 로직이 필요할 때 도입을 고려하여 유지보수성을 증가한다.
• 데이터 처리를 실패했을 때 null을 반환하여 writer에 전달되지 않게 한다.

ItemWriter

Chunk 단위 만큼 한번에 작업을 수행한다.

Chunk 기반 처리 방식

---

Tasklet과 Chunk를 비교하기🤺

구분	Tasklet	Chunk
실행 시점	STEP 실행 중	STEP 실행 전
실행 방식	‘작은 단위’로 분할하여 실행	‘하나의 큰 덩어리’로 실행
커밋 방식	‘Tasklet’ 단위로 처리 후 커밋	‘Chunk’ 단위로 처리 후 커밋
배치 성능	‘작은 단위’의 데이터 처리 시 유리	'대량의 데이터’ 처리 시 유리
재시작	실패한 태스크릿만 다시 실행	실패 시 청크 전체 다시 실행
가독성	코드가 분할되어 가독성 향상	코드가 길어져 가독성 저하
유지보수	수정이 용이	수정이 어려움

---

Parallel Chunk 방식

Parallel Chunk 방식이란 Chunk 방식의 처리에서 더욱 빠른 처리 속도를 위해 Chunk를 독립적으로 처리하여 여러 개의 ‘Chunk를 병렬로 처리’한다. 병렬 처리를 통해 처리 속도를 높일 수 있다는 장점이 있다.

• 예를 들어, 여러 대의 서버에서 동시에 작업을 처리해야 할 때 사용됩니다.

Parallel Chunk 수행 과정

1. 데이터를 적절한 크기로 분할한다.
2. 분할된 각각의 부분을 병렬처리할 수 있도록 작업을 분배한다.
3. 각각의 부분을 병렬처리한다.
4. 처리된 결과를 다시 합친다.

Remote Chunking 방식

Remote Chunking 방식이란 여러 대의 서버에서 대용량 데이터 처리를 수행할 때 사용되며 서버 간에 데이터를 공유하고, 각각의 서버에서 병렬로 처리한다.

---

구현 방법(tasklet 기반 배치)

1. build.gradle에 의존성 추가

//build.gradle
implementation 'org.springframework.boot:spring-boot-starter-batch'
testImplementation 'org.springframework.batch:spring-batch-test'implementation 'org.springframework.boot:spring-boot-starter-batch'

2. BatchConfig 클래스에서 원하는 Job과 Step을 작성한다.

package com.springboot.springbatchtutorial.config;

import lombok.RequiredArgsConstructor;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.DuplicateJobException;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.support.DefaultBatchConfiguration;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.EnableTransactionManagement;

@Configuration
@RequiredArgsConstructor  // Lombok 애노테이션으로 final 또는 @NonNull 필드에 대한 생성자를 자동으로 생성한다.
public class BatchConfig extends DefaultBatchConfiguration {
    @Bean
    public Job testJob(JobRepository jobRepository, PlatformTransactionManager transactionManager) throws DuplicateJobException {
        // Job을 정의하고 시작할 Step을 지정한다.
        // JobBuilder를 사용하여 "testJob"이라는 이름의 잡을 만든다.
        Job job = new JobBuilder("testJob", jobRepository)
                .start(testStep(jobRepository, transactionManager))
                .build();
        return job;
    }

    public Step testStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
        // Step을 정의하고 Tasklet을 지정한다.
        // StepBuilder를 사용하여 "testStep"이라는 이름의 스텝을 만든다.
        Step step = new StepBuilder("testStep", jobRepository)
                .tasklet(testTasklet(), transactionManager) // testTasklet을 tasklet으로 설정하고 트랜잭션 매니저를 사용한다.
                .build();
        return step;
    }

    public Tasklet testTasklet() {
        // Tasklet을 정의하고 비즈니스 로직을 작성
        // Tasklet을 생성합니다.
        return ((contribution, chunkContext) -> {
            System.out.println("***** 10초마다 'Hello batch' 출력!! *****"); // 콘솔에 출력합니다.
            // 원하는 비지니스 로직 작성
            return RepeatStatus.FINISHED; // 작업이 완료되었음을 나타낸다.
        });
    }
}

3. BatchScheduler 클래스에서 @Schedueld 어노테이션을 사용하여 Job의 실행 조건 설정

package com.springboot.springbatchtutorial.scheduler;

import lombok.RequiredArgsConstructor;
import org.springframework.batch.core.*;
import org.springframework.batch.core.configuration.JobRegistry;
import org.springframework.batch.core.configuration.support.JobRegistryBeanPostProcessor;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.NoSuchJobException;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;

@RequiredArgsConstructor // Lombok 애노테이션으로 final 또는 @NonNull 필드에 대한 생성자를 자동으로 생성한다.
@Component // Spring Bean으로 등록됨을 나타낸다.
public class BatchScheduler {

    private final JobLauncher jobLauncher;
    private final JobRegistry jobRegistry;

    @Bean
    public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor(){
        // JobRegistryBeanPostProcessor 빈을 생성한다.
        JobRegistryBeanPostProcessor jobProcessor = new JobRegistryBeanPostProcessor();
        jobProcessor.setJobRegistry(jobRegistry);
        return jobProcessor;
    }

    @Scheduled(cron = "0/10 * * * * *") // 10초마다 실행, // 10초마다 스케줄링하여 실행된다.
    public void runJob() {
        String time = LocalDateTime.now().toString(); // 현재 시간을 문자열로 변환한다.
        try {
            Job job = jobRegistry.getJob("testJob"); // jobRegistry에서 "testJob"을 가져온다.
            JobParametersBuilder jobParam = new JobParametersBuilder().addString("time", time); // JobParametersBuilder에 현재 시간을 추가한다.
            jobLauncher.run(job, jobParam.toJobParameters()); // Job을 실행한다.
        } catch (NoSuchJobException e) {
            throw new RuntimeException(e); // Job을 찾을 수 없는 예외를 처리한다.
        } catch (JobInstanceAlreadyCompleteException |
                 JobExecutionAlreadyRunningException |
                 JobParametersInvalidException |
                 JobRestartException e
        ) {
            throw new RuntimeException(e); // 여러 예외를 처리한다.
        }
    }
}

4. Application 클래스 내 @EnableScheduling 어노테이션 추가하기

package com.springboot.springbatchtutorial;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
public class SpringBatchTutorialApplication {

	public static void main(String[] args) {
		SpringApplication.run(
				SpringBatchTutorialApplication.class, args);
	}

}

application.properties에 설정 코드 추가

# Spring Batch 작업 자동 실행 비활성화
spring.batch.job.enabled=false
# Spring Batch가 시작될 때 스키마를 항상 초기화 (필요에 따라 데이터베이스 테이블을 생성)
spring.batch.jdbc.initialize-schema=always
# 빈 정의 덮어쓰기를 허용 (동일한 이름의 빈을 다른 설정으로 정의할 수 있음)
spring.main.allow-bean-definition-overriding=true

전체적인 실행 흐름🏄🏻‍♂️

1. SpringBatchTutorialApplication 클래스가 애플리케이션을 시작한다.
2. BatchConfig 클래스에서 testJob과 testStep이 정의된다.
3. BatchScheduler 클래스에서 매 10초마다 testJob이 실행되도록 스케줄링된다.
4. testJob이 실행될 때마다 testTasklet이 호출되어 "10초마다 'Hello batch' 출력!!" 메시지를 출력한다.

전체 파일디렉토리 구조

실제 구현한 코드의 실행 결과🔫

---

그럼 어디에 Spring Batch가 사용이 될까?🧐

• 데이터베이스 간의 대량 데이터 전송이나 데이터베이스에서 파일로, 또는 파일에서 데이터베이스로 데이터를 마이그레이션하는 작업에 사용된다.
  예시: A라는 데이터베이스에서 B라는 데이터베이스로 모든 고객 데이터를 이전하는 작업.

• 다양한 소스에서 데이터를 추출(Extract)하고, 필요한 형식으로 변환(Transform)한 후, 최종적으로 타겟 데이터베이스나 파일에 적재(Load)하는 작업을 수행한다.
  예시: 여러 CSV 파일에서 데이터를 추출하고, 특정 비즈니스 로직에 따라 데이터를 변환한 후, 데이터베이스에 적재하는 작업.

• 정기적으로 대량의 데이터를 처리하여 보고서를 생성하는 작업에 사용된다.
  예시: 매월 마지막 날에 지난 달의 매출 데이터를 집계하여 PDF 형식의 보고서를 생성하는 작업.

• 로그 데이터, 트랜잭션 데이터 등의 대량 데이터를 정리하고 집계하는 작업에 사용된다.
  예시: 하루 동안의 웹 서버 로그를 수집하여 방문자 수, 페이지 뷰 수 등을 집계하는 작업.

• 정기적으로 데이터베이스를 백업하거나 복원하는 작업을 자동화할 수 있다.
  예시: 매일 밤 데이터베이스를 백업하고, 백업 파일을 특정 위치에 저장하는 작업.

• 대용량 파일(예: CSV, XML, JSON 등)을 읽고, 처리한 후 결과를 저장하는 작업에 사용된다.
  예시: 수백만 건의 레코드가 포함된 CSV 파일을 읽어 각 레코드를 데이터베이스에 저장하는 작업.

• 정기적으로 데이터베이스의 데이터 정합성을 검증하고, 이상이 있는 데이터를 별도로 처리하는 작업을 수행할 수 있다.
  예시: 매주 데이터베이스의 고객 데이터 정합성을 검증하고, 누락된 데이터를 보고서로 생성하여 관리자에게 알리는 작업.

• 특정 이벤트나 조건에 따라 대량의 이메일을 발송하는 작업을 자동화할 수 있다.
  예시: 특정 프로모션 기간에 이벤트에 참여한 사용자들에게 이메일을 발송하는 작업.

• 대량의 데이터를 분석하고, 머신러닝 모델을 적용하여 예측 결과를 저장하는 작업을 할 수 있다.
  예시: 고객 데이터를 분석하여 고객 이탈을 예측하고, 결과를 데이터베이스에 저장하는 작업.

이와 같이 Spring Boot Batch를 다양한 시나리오에서 사용한다면 반복적으로 수행해야하는 작업이나 대량 데이터 처리 작업을 안정적이고 효율적으로 자동화하기 편할 것이다.

---

Spring Boot 2 vs Spring Boot 3 주의해야 할 점

• 이전 버전까지 @EnableBatchProcessing 를 통해 배치 실행에 필요한 Bean들이 AutoConfiguration 에 의해 등록이 되었다면, 이제는 아무런 영향을 미치지 않는다.
  오히려 있으면 이전과 같이 배치 플로우가 동작을 하지 않는다. SpringBoot 를 이용하고 있고, 이전에 어노테이션으로 BatchRunner 를 생성했었다면, @EnableBatchProcessing 을 빼주어야 한다.

• Ordering 을 통해, Configuration 을 각각 구성하여 하나의 애플리케이션에서 Multiple Job 을 구성한 케이스도 있었는데, 이제는 지원하지 않는다. (1 Application 1 Job 전략)

• JobBuilderFactory, StepBuilderFactory 는 이제 Deprecated 되었다.
  (TO BE) 이제는 다음과 같이 JobBuilder, StepBuilder 를 이용하여 Job, Step 을 구성해야한다.

• Step의 chunk, tasklet 에서 transactionManager 를 인자로 넣어줘야 한다.

여기 기재한 것 이외에도 Spring Boot 3로 넘어오면서 Spring Boot Batch에 많은 변화들이 생겼다. 만약 코드를 작성하고 구동할 때 에러가 발생한다면 찾아보는걸 추천한다.😫

---

참고 문헌

• https://spring.io/batch
• https://docs.spring.io/spring-batch/reference/#domainLanguageOfBatch
• https://docs.spring.io/spring-batch/reference/#job
• https://docs.spring.io/spring-batch/reference/#configureJob
• https://docs.spring.io/spring-batch/reference/#remote-chunking
• https://adjh54.tistory.com/169