Spring Batch란? 간단한 개념과 코드 살펴보기

서론

스프링 배치(Spring Batch)는 대용량 데이터를 처리하기 위한 프레임워크로, 스프링 프레임워크 기반에서 작동합니다. 일반적으로 배치 작업은 대량의 데이터를 처리하거나, 주기적이고 반복적인 작업을 실행하는 데 사용되며, 스프링 배치는 이러한 작업을 효율적이고 안정적으로 처리할 수 있는 대표적으로 아래와 같은 기능들을 제공합니다. 

• 로깅 및 추적
• 트랜잭션 관리
• 작업 처리 통계
• 작업 재시작
• 건너뛰기
• 리소스 관리

 

Batch와 Scheduler의 차이

배치(Batch)는 논리적 또는 물리적으로 관련된 일련의 데이터를 그룹화하여 일괄 처리하는 방법을 의미합니다. 반면에 스케줄러(Scheduler)는 주어진 작업을 미리 정의된 시간에 실행할 수 있게 해주는 도구나 소프트웨어를 의미합니다.

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

Batch 사용 사례

일매출 집계

커머스 사이트에서는 하루에 거래건이 50만에서 100만 건까지 이루어질 수 있습니다. 이런 경우 관련된 데이터는 최소 100만에서 200만 행 이상이 될 수 있는데, 한 달 동안은 이런 데이터가 5000만에서 1억까지 쌓일 수 있습니다. 이런 데이터를 실시간으로 집계하는 쿼리를 실행하면 조회 시간이 길어지고 서버에 많은 부담이 가게 됩니다. 그래서 매일 새벽에 전날의 매출 집계 데이터를 미리 생성해 두어, 외부에서 요청이 오면 전날에 집계한 데이터를 제공하는 방법으로 해결합니다. 

https://jojoldu.tistory.com/324?category=635883

구독 서비스 

정해진 시간에 구독자들에게 메일을 일괄 전송하는 경우, 배치 처리를 활용하면 서비스 구현이 간단해집니다. 전송할 데이터 내역과 구독자 정보를 활용하여 구독을 신청한 회원들에게 규칙적으로 메일을 보낼 수 있습니다.

데이터 백업

대규모 데이터베이스를 운영하게 되면, 데이터의 일관성을 보장하기 위해 주기적인 데이터 백업이 필수적입니다. 그러나 이런 백업 작업은 시스템에 상당한 부담을 줄 수 있기에 일반적으로 사용자 트래픽이 상대적으로 적은 시간대에 배치 처리 방식을 통해 백업 작업을 실행합니다.

 

Spring Batch의 용어

스프링 배치에서 사용하는 주요 용어들과 의미는 다음과 같습니다.

• Job
  • Job은 전체 배치 처리 과정을 추상화한 개념으로, 하나 또는 그 이상의 Step을 포함하며, 스프링 배치 계층에서 가장 상위에 위치합니다.
  • 각 Job은 고유한 이름을 가지며, 이 이름은 실행에 필요한 파라미터와 함께 JobInstance를 구별하는 데 사용됩니다.
• JobInstance
  • JobInstance는 특정 Job의 실제 실행 인스턴스를 의미합니다. 예를 들어, "매일 아침 8시에 데이터를 처리"하는 Job을 구성한다고 가정하면, 1월 1일, 1월 2일 등 매일 실행될 때마다 새로운 JobInstance가 생성됩니다.
  • 한번 생성된 JobInstance는 해당 날짜의 데이터를 처리하는 데 사용되며, 실패했을 경우 같은 JobInstance를 다시 실행하여 작업을 완료할 수 있습니다.
• JobParameters
  • JobParameters는 JobInstance를 생성하고 구별하는 데 사용되는 파라미터입니다.
  • Job이 실행될 때 필요한 파라미터를 공하며, JobInstance를 구별하는 역할도 합니다.
  • 스프링 배치는 String, Double, Long, Date 이렇게 4가지 타입의 파라미터를 지원합니다.
• JobExecution
  • JobExecution은 JobInstance의 한 번의 시행 시도를 나타냅니다.
  • 예를 들어, 1월 1일에 실행된 JobInstance가 실패했을 때 재시도하면, 같은 JobInstance에 대한 새로운 JobExcecution이 생성됩니다.
  • JobExecution은 실행 상태, 시작시간, 종료시간, 생성시간 등 JobInstance의 실행에 대한 세부 정보를 담고 있습니다.
• Step
  • Step은 Job의 하위 단계로서 실제 배치 처리 작업이 이루어지는 단위입니다.
  • 한 개 이상의 Step으로 Job이 구성되며, 각 Step은 순차적으로 처리됩니다.
  • 각 Step 내부에서는 ItemReader, ItemProcessor, ItemWriter를 사용하는 chunk 방식 또는 Tasklet 하나를 가질 수 있습니다.
• StepExecution
  • StepExecution은 Step의 한 번의 실행을 나타내며, Step의 실행 상태, 실행 시간 등의 정보를 포함합니다.
  • JobExecution과 유사하게, 각 Step의 실행 시도마다 새로운 StepExecution이 생성됩니다.
  • 또한, 읽은 아이템의 수, 쓴 아이템의 수, 커밋 횟수, 스킵한 아이템의 수 등의 Step 실행에 대한 상세 정보도 포합 합니다.
• ExecutionContext
  • ExecutionContext는 Step 간 또는 Job 실행 도중 데이터를 공유하는 데 사용되는 저장소입니다.
  • JobExecutionContext와 StepExecutionContext 두 종류가 있으며, 범위와 저장 시점에 따라 적절하게 사용됩니다.
  • Job이나 Step이 실패했을 경우, ExecutionContext를 통해 마지막 실행 상태를 재구성하여 재시도 또는 복구 작업을 수행할 수 있습니다.
• JobRepository
  • JobRepository는 배치 작업에 관련된 모든 정보를 저장하고 관리하는 메커니즘입니다.
  • Job 실행정보(JobExecution), Step 실행정보(StepExecution), Job 파라미터(JobParameters)등을 저장하고 관리합니다.
  • Job이 실행될 때, JobRepository는 새로운 JobExecution과 StepExecution을 생성하고, 이를 통해 실행 상태를 추적합니다.
• JobLauncher
  • JobLauncher는 Job과 JobParameters를 받아 Job을 실행하는 역할을 합니다.
  • 이는 전반적인 Job의 생명 주기를 관리하며, JobRepository를 통해 실행 상태를 유지합니다.
• ItemReader
  • ItemReader는 배치 작업에서 처리할 아이템을 읽어오는 역할을 합니다.
  • 여러 형식의 데이터 소스(예: 데이터베이스, 파일, 메시지 큐 등)로 부터 데이터를 읽어오는 다양한 ItemReader 구현체가 제공됩니다.
• ItemProcessor
  • ItemProcessor는 ItemReader로부터 읽어온 아이템을 처리하는 역할을 합니다.
  • 이는 선택적인 부분으로서, 필요에 따라 사용할 수 있으며, 데이터 필터링, 변환 등의 작업을 수행할 수 있습니다.
• ItemWriter
  • ItemWriter는 ItemProcessor에서 처리된 데이터를 최종적으로 기록하는 역할을 합니다.
  • ItemWriter 역시 다양한 형태의 구현체를 통해 데이터베이스에 기록하거나, 파일을 생성하거나 메시지를 발행하는 등 다양한 방식으로 데이터를 쓸 수 있습니다.
• Tasklet
  • Tasklet은 간단한 단일 작업, 예를 들어 리소스의 정리 또는 시스템 상태의 체크 등을 수행할 때 사용됩니다.
  • 이는 스프링 배치의 Step 내에서 단일 작업을 수행하기 위한 인터페이스로, 일반적으로 ItemReader, ItemProcessor, ItemWriter의 묶음을 가지는 Chunk 기반 처리 방식과는 다릅니다.
  • Tasklet의 execute 메서드는 Step의 모든 처리가 끝날 때까지 계속 호출됩니다.
• JobOperator
  • JobOperator는 외부 인터페이스로, Job의 실행과 중지, 재시작 등의 배치 작업 흐름제어를 담당합니다.
  • 이 인터페이스를 통해 JobLauncher와 JobRepository에 대한 직접적인 접근 없이도 배치 작업을 수행하고 상태를 조회할 수 있습니다.
• JobExplorer
  • JobExplorer는 Job의 실행 이력을 조회하는 데 사용됩니다.
  • JobRepository에서 제공하는 정보와 유사하지만, JobRepository는 주로 Job의 실행 도중인 상태에 대해 업데이트하고 관리하는 반면, JobExplorer는 주로 읽기 전용 접근에 초점을 맞추고 있습니다.

 

메타 테이블

먼저 스프링 배치는 배치 작업의 상태를 관리하기 위한 메타 데이터를 저장하는 아래 6개의 테이블들을 자동으로 생성합니다. 또한 배치 작업을 수행하면 자동으로 생성된 테이블들의 컬럼 값들이 채워집니다.

https://docs.spring.io/spring-batch/docs/current/reference/html/schema-appendix.html

1. BATCH_JOB_INSTANCE

CREATE TABLE BATCH_JOB_INSTANCE  (
  JOB_INSTANCE_ID BIGINT  PRIMARY KEY ,
  VERSION BIGINT,
  JOB_NAME VARCHAR(100) NOT NULL ,
  JOB_KEY VARCHAR(32) NOT NULL
);

BATCH_JOB_INSTANCE 테이블은 JobInstance와 관련된 모든 정보를 가지며, 전체 계층 구조의 최상위 역할을 합니다.

• JOB_INSTANCE_ID: 실행된 JobInstance의 ID
• VERSION: 배치 테이블의 낙관적 락 전략을 위해 사용되는 값
• JOB_NAME: 실행된 Job의 이름으로, null이 아니 여야함
• JOB_KEY: JobParameter로 생성된 JobInstance의 키(Job 중복 수행 체크를 위한 고유키)

2. BATCH_JOB_EXECUTION

CREATE TABLE BATCH_JOB_EXECUTION  (
  JOB_EXECUTION_ID BIGINT  PRIMARY KEY ,
  VERSION BIGINT,
  JOB_INSTANCE_ID BIGINT NOT NULL,
  CREATE_TIME TIMESTAMP NOT NULL,
  START_TIME TIMESTAMP DEFAULT NULL,
  END_TIME TIMESTAMP DEFAULT NULL,
  STATUS VARCHAR(10),
  EXIT_CODE VARCHAR(20),
  EXIT_MESSAGE VARCHAR(2500),
  LAST_UPDATED TIMESTAMP,
  constraint JOB_INSTANCE_EXECUTION_FK foreign key (JOB_INSTANCE_ID)
  references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
) ;

BATCH_JOB_EXECUTION 테이블은 JobExecution와 관련된 모든 정보가 들어있습니다. JobExcution은 JobInstance가 실행될 때마다 시작시간, 종료시간, 종료코드 등 다양한 정보를 가지고 있습니다.

• JOB_EXECUTION_ID: 실행된 JobExecution의 ID
• VERSION: 배치 테이블의 낙관적 락 전략을 위해 사용되는 값
• JOB_INSTANCE_ID: BATCH_JOB_INSTANCE 테이블의 외래 키
• CREATE_TIME: JobExecution이 생성된 시간
• START_TIME: JobExecution이 실행된 시간
• END_TIME: JobExecution이 종료된 시간(성공과 실패 여부에 상관없이 실행이 완료된 시간을 의미)
• STATUS: JobExecution의 상태(BatchStatus의 Enum 타입)
• EXIT_CODE: JobExecution의 종료 코드
• EXIT_MESSAGE: JobExecution의 종료 메시지. 에러가 발생한 경우 에러메시지
• LAST_UPDATED: JobExcution이 수정된 시간

3. BATCH_JOB_EXECUTION_PARAMS

CREATE TABLE BATCH_JOB_EXECUTION_PARAMS  (
	JOB_EXECUTION_ID BIGINT NOT NULL ,
	PARAMETER_NAME VARCHAR(100) NOT NULL ,
	PARAMETER_TYPE VARCHAR(100) NOT NULL ,
	PARAMETER_VALUE VARCHAR(2500) ,
	IDENTIFYING CHAR(1) NOT NULL ,
	constraint JOB_EXEC_PARAMS_FK foreign key (JOB_EXECUTION_ID)
	references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
);

BATCH_JOB_EXECUTION_PARAMS 테이블은 JobParameters와 관련된 모든 정보가 들어있습니다.

• JOB_EXECUTION_ID: 실행된 JobExecution의 ID(BATCH_JOB_EXECUTION 테이블의 외래 키)
• TYPE_CD: JobPameter의 타입 코드(string, date, long, double)
• KEY_NAME: JobPameter의 이름
• STRING_VAL: JobPameter의 String값(TYPE_CD가 String일 때 값이 존재)
• DATETIME: JobPameter의 Date값(TYPE_CD가 Date 일 때 값이 존재)
• LONG_VAL: JobPameter의 Long값. (TYPE_CD가 Long일 때 값이 존재)
• DOUBLE_VAL: JobPameter의 Double값. (TYPE_CD가 Double일 때 값이 존재)
• IDENTIFYING: JobInstance의 키의 생성에 포함되었는지 여부

4. BATCH_JOB_EXECUTION_CONTEXT

CREATE TABLE BATCH_JOB_EXECUTION_CONTEXT  (
  JOB_EXECUTION_ID BIGINT PRIMARY KEY,
  SHORT_CONTEXT VARCHAR(2500) NOT NULL,
  SERIALIZED_CONTEXT CLOB,
  constraint JOB_EXEC_CTX_FK foreign key (JOB_EXECUTION_ID)
  references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
) ;

BATCH_JOB_EXECUTION_CONTEXT 테이블은 작업의 실행 컨텍스트와 관련된 모든 정보가 들어있습니다. 각 JobExecution마다 정확히 하나의 JobExecutionContext가 있습니다. 이 ExecutionContext 데이터는 일반적으로 JobInstance가 실패 시 중단된 위치에서 다시 시작할 수 있는 정보를 저장하고 있습니다.

• JOB_EXECUTION_ID: 실행된 JobExecution의 ID
• SHORT_CONTEXT: 문자열로 저장된 JobExecutionContext 정보
• SERIALIZED_CONTEXT: 직렬화하여 저장된 JobExecutionContext 정보

5. BATCH_STEP_EXECUTION

CREATE TABLE BATCH_STEP_EXECUTION  (
  STEP_EXECUTION_ID BIGINT NOT NULL PRIMARY KEY ,
  VERSION BIGINT NOT NULL,
  STEP_NAME VARCHAR(100) NOT NULL,
  JOB_EXECUTION_ID BIGINT NOT NULL,
  CREATE_TIME TIMESTAMP NOT NULL,
  START_TIME TIMESTAMP DEFAULT NULL ,
  END_TIME TIMESTAMP DEFAULT NULL,
  STATUS VARCHAR(10),
  COMMIT_COUNT BIGINT ,
  READ_COUNT BIGINT ,
  FILTER_COUNT BIGINT ,
  WRITE_COUNT BIGINT ,
  READ_SKIP_COUNT BIGINT ,
  WRITE_SKIP_COUNT BIGINT ,
  PROCESS_SKIP_COUNT BIGINT ,
  ROLLBACK_COUNT BIGINT ,
  EXIT_CODE VARCHAR(20) ,
  EXIT_MESSAGE VARCHAR(2500) ,
  LAST_UPDATED TIMESTAMP,
  constraint JOB_EXECUTION_STEP_FK foreign key (JOB_EXECUTION_ID)
  references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
) ;

BATCH_STEP_EXECUTION 테이블은 StepExecution 객체와 관련된 모든 정보가 저장됩니다. BATCH_JOB_EXECUTION 테이블과 유사하며, 생성된 각 JobExecution에 대해 항상 단계당 하나 이상의 항목이 존재합니다. STEP의 EXECUTION 정보인 읽은 수, 커밋 수, 스킵 수 등 다양한 정보를 추가로 담고 있습니다.

• STEP_EXECUTION_ID: 실행된 StepExecution의 ID
• VERSION: 배치 테이블의 낙관적 락 전략을 위해 사용되는 값
• STEP_NAME: 실행된 StepExecution의 Step 이름
• JOB_EXECUTION_ID: 실행된 JobExecution의 ID
• START_TIME: StepExecution이 시작된 시간
• END_TIME: StepExecution이 종료된 시간(성공과 실패 여부에 상관없이 실행이 완료된 시간을 의미)
• STATUS: StepExecution의 상태(BatchStatus의 Enum 타입)
• COMMIT_COUNT: StepExecution 실행 중, 커밋한 횟수
• READ_COUNT: StepExecution 실행 중, 읽은 데이터 수
• FILTER_COUNT: StepExecution 실행 중, 필터링된 데이터 수
• WRITE_COUNT: StepExecution 실행 중, 작성 및 커밋된 데이터 수
• READ_SKIP_COUNT: StepExecution 실행 중, 읽기를 스킵한 데이터 수
• WRITE_SKIP_COUNT: StepExecution 실행 중, 작성을 스킵한 데이터 수
• PROCESS_SKIP_COUNT: StepExecution 실행 중, 처리를 스킵한 데이터 수
• EXIT_CODE: StepExecution의 종료 코드
• ROLLBACK_COUNT: StepExecution 실행 중, 롤백 횟수
• EXIT_MESSAGE: StepExecution의 종료 메시지. 에러가 발생한 경우 에러메시지
• LAST_UPDATED: StepExecution이 수정된 시간

6. BATCH_STEP_EXECUTION_CONTEXT

CREATE TABLE BATCH_JOB_EXECUTION_CONTEXT  (
  JOB_EXECUTION_ID BIGINT PRIMARY KEY,
  SHORT_CONTEXT VARCHAR(2500) NOT NULL,
  SERIALIZED_CONTEXT CLOB,
  constraint JOB_EXEC_CTX_FK foreign key (JOB_EXECUTION_ID)
  references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
) ;

BATCH_STEP_EXECUTION_CONTEXT 테이블은 StepExecutionContext와 관련된 모든 정보가 저장되며, 스텝 실행당 정확히 하나의 ExecutionContext가 있으며, 특정 스텝 실행을 위해 유지되어야 하는 모든 데이터가 포함되어 있습니다. 이 ExecutionContext 데이터는 일반적으로 JobInstance가 실패 시 중단된 위치에서 다시 시작할 수 있는 정보를 저장하고 있습니다.

• STEP_EXECUTION_ID: 실행된 StepExecution의 ID
• SHORT_CONTEXT: 문자열로 저장된 StepExecutionContext 정보
• SERIALIZED_CONTEXT: 직렬화하여 저장된 StepExecutionContext 정보

각 메타 테이블들에 값이 들어가는 것을 확인해보고 싶으신 분은 https://jojoldu.tistory.com/326를 참고하면 좋을 것 같습니다.
 

Spring Batch 사용하기

Spring Batch에서 각 Job은 여러 Step의 모음으로 구성되며, 각 Step은 특정한 비즈니스 로직 처리 단위를 나타냅니다.

1. 단일 Step 구성하기

@Slf4j
@Configuration
@EnableBatchProcessing
public class SimpleJobConfig {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Autowired
    public SimpleJobConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
    }

    @Bean
    public Job exampleJob() {
        return jobBuilderFactory.get("exampleJob")
                .start(step())
                .build();
    }

    @Bean
    public Step step() {
        return stepBuilderFactory.get("step")
                .tasklet((contribution, chunkContext) -> {
                    log.info("Performing step!");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }
}

위 코드에서는 JobBuilderFactory와 StepBuilderFactory를 사용하여 Job과 Step을 생성합니다. exampleJob이라는 Job은 단일 step만을 포함하며, 이 step에서는 tasklet을 통해 특정한 동작을 수행합니다. 그리고 RepeatStatus.FINISHED를 반환하여 작업이 성공적으로 완료되었음을 나타냅니다.

2. 다중 Step 구성하기

Sequential Flow

위 이미지처럼 Step을 순차적으로 실행할 수 있습니다.

@Slf4j
@Configuration
@EnableBatchProcessing
public class MultiStepJobConfig {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Autowired
    public MultiStepJobConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
    }

    @Bean
    public Job exampleJob() {
        return jobBuilderFactory.get("exampleJob")
                .start(startStep())
                .next(nextStep())
                .next(lastStep())
                .build();
    }

    @Bean
    public Step startStep() {
        return stepBuilderFactory.get("startStep")
                .tasklet((contribution, chunkContext) -> {
                    log.info("Start Step!");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    @Bean
    public Step nextStep() {
        return stepBuilderFactory.get("nextStep")
                .tasklet((contribution, chunkContext) -> {
                    log.info("Next Step!");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    @Bean
    public Step lastStep() {
        return stepBuilderFactory.get("lastStep")
                .tasklet((contribution, chunkContext) -> {
                    log.info("Last Step!!");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }
}

exampleJob이라는 Job은 startStep, nextStep, lastStep의 세 개의 Step을 순차적으로 수행합니다. start() 메서드로 최초 실행될 Step을 설정하고, next() 메서드로 그다음에 수행될 Step을 연결합니다.

3. Flow를 통한 Step 구성하기

Conditional Flow

위 이미지처럼 이전 Step의 성공 여부에 따라 분기처리를 할 수 있습니다.

@Slf4j
@Configuration
@EnableBatchProcessing
public class FlowJobConfig {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Autowired
    public FlowJobConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
    }

    @Bean
    public Job exampleJob() {
        // Job 생성
        return jobBuilderFactory.get("exampleJob")
            .start(startStep())
                .on(ExitStatus.FAILED.getExitCode()) // startStep의 ExitStatus가 FAILED일 경우
                .to(failOverStep()) // failOver Step을 실행 시킨다.
                .on("*") // failOver Step의 결과와 상관없이
                .to(writeStep()) // write Step을 실행 시킨다.
                .end() // Flow를 종료시킨다.
            .from(startStep()) // startStep이 FAILED가 아니고 COMPLETED일 경우
                .on(ExitStatus.COMPLETED.getExitCode())
                .to(processStep()) // process Step을 실행 시킨다
                .on("*") // process Step의 결과와 상관없이
                .to(writeStep()) // write Step을 실행 시킨다.
                .end() // Flow를 종료 시킨다.
            .from(startStep()) // startStep의 결과가 FAILED, COMPLETED가 아닌 모든 경우
                .on("*")
                .to(writeStep()) // write Step을 실행시킨다.
                .on("*") // write Step의 결과와 상관없이
                .end() // Flow를 종료시킨다.
            .end()
            .build();
    }

    @Bean
    public Step startStep() {
        // 첫번째 Step 생성
        return stepBuilderFactory.get("startStep")
            .tasklet((contribution, chunkContext) -> {
                log.info("Start Step!");

                String result = "COMPLETED";
                // String result = "FAIL";
                // String result = "UNKNOWN";

                // Flow에서 on은 RepeatStatus가 아닌 ExitStatus를 바라본다.
                if ("COMPLETED".equals(result)) 
                    contribution.setExitStatus(ExitStatus.COMPLETED);
                else if ("FAIL".equals(result))  
                    contribution.setExitStatus(ExitStatus.FAILED);
                else if ("UNKNOWN".equals(result))  
                    contribution.setExitStatus(ExitStatus.UNKNOWN);

                return RepeatStatus.FINISHED;
            })
            .build();
    }

    @Bean
    public Step failOverStep() {
        // 실패 시 수행할 Step 생성
        return stepBuilderFactory.get("failOverStep")
            .tasklet((contribution, chunkContext) -> {
                log.info("FailOver Step!");
                return RepeatStatus.FINISHED;
            })
            .build();
    }

    @Bean
    public Step processStep() {
        // 처리를 위한 Step 생성
        return stepBuilderFactory.get("processStep")
            .tasklet((contribution, chunkContext) -> {
                log.info("Process Step!");
                return RepeatStatus.FINISHED;
            })
            .build();
    }

    @Bean
    public Step writeStep() {
        // 결과를 기록하기 위한 Step 생성
        return stepBuilderFactory.get("writeStep")
            .tasklet((contribution, chunkContext) -> {
                log.info("Write Step!");
                return RepeatStatus.FINISHED;
            })
            .build();
    }
}

위 코드에서는 다음과 같은 구조를 가진 Job을 정의하고 있습니다.

• startStep(): 시작 Step을 정의하고 있습니다. ExitStatus를 COMPLETED, FAILED, UNKNOWN 중 하나로 설정하여 다음 Step의 수행을 제어합니다.
• failOverStep(): startStep()의 ExitStatus가 FAILED일 경우 수행될 Step입니다.
• processStep(): startStep()의 ExitStatus가 COMPLETED일 경우 수행될 Step입니다.
• writeStep(): 위의 모든 상황 후에 수행될 Step입니다.

각 Step은 Tasklet을 통해 실제 수행될 로직을 정의하고 있으며, on() 메서드를 통해 ExitStatus에 따라 수행될 Step을 지정하고 end()를 통해 Flow를 종료합니다. 
 

STEP을 구성하는 Tasklet과 Chunk 지향 처리

Job은 Step들을 순차적으로 수행하게 되는데, Step은 주로 Tasklet 방식과 chunk 방식이 존재합니다.

https://www.codeusingjava.com/boot/batch/hello

* Tasklet 방식

Tasklet은 기본적으로 하나의 작업을 수행하는 방식입니다. 대체로 단순하거나 복잡하지 않은 작업을 수행하는 데 적합하며, 전체 데이터를 처리하는 것이 아니라 일부 데이터나 단일 작업을 처리하는 데 주로 사용됩니다.

Tasklet 방식의 작업 흐름은 아래와 같습니다.

• Tasklet 인터페이스의 execute() 메서드를 구현하여 사용한다. 이 메서드는 하나의 트랜잭션 범위에서 실행된다.
• execute() 메서드는 RepeatStatus를 반환하는데 이는 Tasklet의 실행 상태를 나타낸다.
• RepeatStatus.FINISHED를 반환하면, 해당 Tasklet의 처리가 완료된 것을 의미한다.
• RepeatStatus.CONTINUABLE를 반환하면, Tasklet이 계속 실행되어야 함을 의미한다.

@Configuration
public class TaskletStepConfiguration {

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public Step taskletStep() {
        return stepBuilderFactory.get("taskletStep")
                .tasklet((contribution, chunkContext) -> {
                    // 여기에 Tasklet 로직 작성
                    System.out.println("Tasklet step executed");

                    return RepeatStatus.FINISHED;
                })
                .build();
    }
}

Tasklet방식은 작업의 단순성과 명확성 때문에 특정 상황에서 유용하게 사용될 수 있습니다. 그러나 복잡하거나 대량의 데이터 처리를 요구하는 경우에는 Chunk 방식이 더 적합합니다.

* Chunk 방식

Chunk 방식은 대용량 데이터를 효과적으로 처리하기 위해 사용합니다. 큰 데이터를 일련의 작은 데이터 묶음(Chunk)으로 나누고, 각 Chunk를 개별적인 트랜잭션 범위 내에서 처리하는 방식을 취합니다.

Chunk방식의 작업 흐름은 아래와 같습니다.

• 각 Chunk 처리는 Reader, Processor, Writer 세 단계로 구성된다.
• Reader는 데이터 소스로부터 데이터를 읽어와서 Chunk를 생성한다. 이 데이터는 일반적으로 데이터베이스나, 파일, 또는 메시지 큐 등이 될 수 있다.
• Processor는 읽어온 데이터에 대해 필요한 처리를 수행한다. 이 처리는 데이터 검증, 필터링, 변환 등 다양한 형태를 가질 수 있다.
• Writer는 처리된 데이터를 최종적으로 저장한다.

간단한 예시 코드는 다음과 같습니다.

@Configuration
public class ChunkStepConfiguration {

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public ItemReader<String> reader() {
        return new ListItemReader<>(Arrays.asList("Ahn", "Ju", "Hyeong"));
    }

    @Bean
    public ItemProcessor<String, String> processor() {
        return item -> item.toUpperCase();
    }

    @Bean
    public ItemWriter<String> writer() {
        return items -> items.forEach(System.out::println);
    }

    @Bean
    public Step chunkStep() {
        return stepBuilderFactory.get("chunkStep")
                .<String, String>chunk(10)
                .reader(reader())
                .processor(processor())
                .writer(writer())
                .build();
    }
}

위 코드에서 ItemReader는 "Ahn", "Ju", "Hyeong"이라는 문자열 리스트를 읽어와서 ItemProcessor에서 각 문자열을 대문자로 변환하여 ItemWriter에서 각 항목을 콘솔에 출력합니다.

좀 더 세부적인 JPA 환경에서 사용했을 때의 예시 코드는 다음과 같습니다.

@Configuration
@EnableBatchProcessing
public class BookOrderJobConfiguration {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private EntityManagerFactory entityManagerFactory;

    // Job 설정
    @Bean
    @JobScope
    public Job bookOrderJob() throws Exception {
        return jobBuilderFactory.get("bookOrderJob")
                .start(bookOrderStep()) // "bookOrderStep"이라는 이름의 Step을 시작점으로 설정
                .build();
    }

    // Step 설정
    @Bean
    @StepScope
    public Step bookOrderStep() throws Exception {
        return stepBuilderFactory.get("bookOrderStep")
                .<Book, Order>chunk(10) // 한 번에 처리할 데이터 항목의 크기(chunk size)를 10으로 설정
                .reader(bookReader()) // Reader 설정
                .processor(bookOrderProcessor()) // Processor 설정
                .writer(orderWriter()) // Writer 설정
                .build();
    }

    // Reader 설정 - 데이터베이스로부터 특정 조건에 맞는 데이터를 읽어오는 역할
    @Bean
    @StepScope
    public JpaPagingItemReader<Book> bookReader() throws Exception {
        Map<String, Object> parameterValues = new HashMap<>();
        parameterValues.put("stock", 5); // stock 파라미터 값을 5로 설정

        return new JpaPagingItemReaderBuilder<Book>()
                .name("JpaPagingItemReader")
                .entityManagerFactory(entityManagerFactory)
                .queryString("SELECT b FROM Book b WHERE b.stock <= :stock ORDER BY b.id ASC") // 재고가 5 이하인 책을 조회하는 쿼리
                .parameterValues(parameterValues)
                .pageSize(10) // 페이징 사이즈를 10으로 설정
                .build();
    }

    // Processor 설정 - 읽어온 데이터를 기반으로 비즈니스 로직을 수행하는 역할
    @Bean
    @StepScope
    public ItemProcessor<Book, Order> bookOrderProcessor() {
        return book -> {
            Order order = new Order();
            order.setBook(book);
            order.setQuantity(10);  // Reorder quantity
            return order; // 재고가 부족한 책에 대해 재주문을 생성
        };
    }

    // Writer 설정 - 처리된 결과 데이터를 데이터베이스에 저장하는 역할
    @Bean
    @StepScope
    public JpaItemWriter<Order> orderWriter() {
        return new JpaItemWriterBuilder<Order>()
                .entityManagerFactory(entityManagerFactory)
                .build();
    }
}

위 코드는 재고가 특정 수량 이하인 Book에 대해서 재주문(Order)을 생성하는 작업을 수행하는 코드입니다. 재주문 작업은 bookOrderJob이라는 Job으로 구성되며, bookOrderStep이라는 단일 Step을 가집니다.

Paging Size와 Chunk Size 란?

Spring Batch는 대용량 데이터를 효과적으로 처리하기 위해 ItemReader와 ItemWriter를 제공하는데 이 중 Paging 처리는 대용량 데이터를 효율적으로 읽어오는 방법 중 하나입니다.

• Paging은 데이터를 일정한 크기의 페이지로 분할하여 처리하는 것을 의미하는데, 예를 들어 만약 데이터가 1000개가 있고 페이지 크기를 100으로 설정한다면, 총 10페이지로 데이터를 나누어 처리하게 됩니다.
• Chunk는 Spring Batch에서 트랜잭션 범위를 설정하는 방법 중 하나입니다. Chunk size는 한 번에 처리(커밋)될 데이터 항목의 수를 의미하며, 만약 Chunk size를 10으로 설정한다면, 각 트랜잭션은 10개의 데이터 항목을 처리하게 됩니다.

Paging Size와 Chunk Size의 관계

Paging Size가 5이고, Chunk Size가 10일 경우, 2번의 Read가 이루어진 후에 1번의 Transaction이 수행됩니다. 이는 한 번의 Transaction을 위해 2번의 쿼리 수행이 발생하게 되므로, 이러한 상황을 효율적이지 않습니다.

따라서 효과적인 성능 향상을 위해선 Spring Batch에서 권장하는 것처럼 페이지 크기를 상당히 크게 설정하고 페이지 크기와 일치하는 커밋 간격(Chunk Size)을 사용하는 것이 좋습니다. 즉 Paging Size와 Chunk Size을 같은 값으로 설정하는 것입니다.

PagingReader 사용 시 주의사항

페이징 처리 시 각 쿼리에 Offset과 Limit를 지정해 주어야 하는데 이는 PageSize를 지정하면 Batch에서 Offset과 Limit를 지정해 줍니다. 하지만 페이징 처리를 할 때마다 새로운 쿼리를 실행하기 때문에 데이터 순서가 보장될 수 있도록 반드시 Order By를 사용하여야 합니다.
 

Batch 테스트 코드

@RunWith(SpringRunner.class)
@SpringBootTest
public class SampleBatchJobTest {

    // JobLauncherTestUtils는 Spring Batch가 제공하는 유틸리티 클래스로, 
    // Job을 실행하고 그 결과를 테스트하는데 사용됨.
    @Autowired
    private JobLauncherTestUtils jobLauncherTestUtils;

    // JobRepositoryTestUtils는 Job 실행에 관련된 메타데이터를 제거하는데 사용,
    // 이를 통해 각 테스트가 독립적으로 실행될 수 있도록 보장함.
    @Autowired
    private JobRepositoryTestUtils jobRepositoryTestUtils;

    @Before
    public void clearMetaData() {
        jobRepositoryTestUtils.removeJobExecutions();
    }

    @Test
    public void testJob() throws Exception {
        // Job을 실행하고 그 결과를 JobExecution 객체로 받음.
        JobExecution jobExecution = jobLauncherTestUtils.launchJob();

        // 실행된 Job의 상태가 COMPLETED(성공적으로 완료됨)인지 확인
        assertThat(jobExecution.getStatus()).isEqualTo(BatchStatus.COMPLETED);
    }
}

기본적으로 위와 같이 작성할 수 있습니다. 
 

참고

• https://nankisu.tistory.com/68
• https://docs.spring.io/spring-batch/docs/current/reference/html/schema-appendix.html#metaDataSchema
• https://khj93.tistory.com/entry/Spring-Batch%EB%9E%80-%EC%9D%B4%ED%95%B4%ED%95%98%EA%B3%A0-%EC%82%AC%EC%9A%A9%ED%95%98%EA%B8%B0