스프링 배치(Spring Batch) 적용하기

Reference

1. https://www.youtube.com/watch?v=MNzPsOQ3NJk&list=PLJkjrxxiBSFCaxkvfuZaK5FzqQWJwmTfR

2. https://tech.kakaopay.com/post/ifkakao2022-batch-performance-read/

3. https://dkswnkk.tistory.com/707

스프링 배치(Spring Batch)

• 대규모 데이터 처리를 위한 프레임워크로, 배치 작업을 효율적으로 수행할 수 있게 도와줌

• 대량의 데이터를 특정 작업 단위로 묶어서 일괄 처리하는 것을 의미함

• 배치와 스케줄러의 관계 ?
• 배치는 데이터를 그룹화하여 일괄 처리하는 것이고, 스케줄러는 일정 시간마다 자동으로 실행시켜 주는 것
• 스케줄러를 사용하여, 배치 작업을 정해진 시간에, 또는 일정 주기마다 반복적으로 실행할 수 있도록 해줄 수 있다

• 주요 기능
1. Chunk 기반 처리
• 데이터를 일정 크기(Chunk)로 나누어 처리하는 방식으로, 메모리 사용을 최적화하고, 대규모 데이터를 효율적으로 처리할 수 있다
• Chunk로 분리하여 처리를 함으로써
• 메모리 효율성을 늘릴 수 있음 → 모든 데이터를 한번에 처리하면 메모리 과부하
• 트랜잭션 관리 → 중간에 장애가 나도 전체 작업의 일부만 처리하면 됨
• I/O 작업 횟수를 줄여 성능 최적화
2. 병렬 처리
• 멀티스레드, 멀티프로세스 방식으로 작업을 병렬로 처리하여 성능을 높일 수 있음
3. 재시작 및 리트라이
• 실패한 배치 작업을 중간부터 재시작하거나 특정 Step을 다시 시도할 수 있음
4. Job 파라미터
• 배치 작업 실행 시 외부에서 파라미터를 받아 작업에 반영할 수 있음

• 주요 특징
1. 대용량 데이터 처리
• 데이터베이스, 파일 시스템, 큐 등 다양한 데이터 소스에서 대규모 데이터를 읽고 처리한 후, 결과를 다시 저장하는 작업에 최적화됨
2. 재사용 가능
• 배치 작업의 구성 요소를 모듈화하여 여러 작업에 재사용할 수 있도록 설계됨
3. 내장된 관리 기능
• 트랜잭션 관리, 로깅/추적, 작업 상태 저장 등 배치 처리에서 필요한 다양한 관리 기능을 제공
4. 확장 가능성
• 확장 가능한 아키텍처를 제공하여 다양한 요구사항에 맞게 커스터마이징 가능
 

주요 구성 요소 및 코드 진행

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

1. JobRepository
• 스프링 배치에서 배치 작업의 실행 상태와 메타데이터를 저장하고 관리하는 핵심 구성 요소
• 작업 메타데이터 저장, 트랜잭션 관리, 재시작/복구 지원, 배치 작업 상태 추적 등을 담당한다
• application.yaml에서 다음 설정을 통해 JobRepository를 정의한다.

spring.batch.job.enabled=false
spring.batch.jdbc.initialize-schema=always
spring.batch.jdbc.schema=classpath:org/springframework/batch/core/schema-mysql.sql

2. Job
• 배치 작업의 최상위 개념으로, 하나의 배치 작업을 의미
• Job은 여러 개의 Step으로 구성되며, 각 Step이 순차적으로 또는 조건에 따라 실행
• Job은 배치 작업의 실행 단위로, 보통 하나의 Job이 하나의 비즈니스 프로세스를 나타냄
• Job의 실행 상태와 이력을 추적하기 위해 메타데이터를 관리함. 실행 결과는 성공, 실패, 중단 등으로 기록

    @Bean
    public Job firstJob() {
        System.out.println("First job");
        return new JobBuilder("firstJob", jobRepository )
                .start(fisrtStep())
                .build();
    }

3. Step
• Job을 구성하는 하나의 단계로, 각 Step은 독립적으로 실행되며, 특정 작업(데이터 읽기, 처리, 쓰기 등)을 수행
• Step은 ItemReader, ItemProcessor, ItemWriter와 같은 구성 요소들을 포함하며, 이를 통해 데이터 처리 작업을 수행함(Chunk 처리)
• Chunk가 아닌 Tasklet 처리 방식도 있지만, 이건 매우 간단한 동작만 가능하기 때문에 거의 사용 x
• 각 Step은 트랜잭션 단위로 실행될 수 있으며, 트랜잭션 관리, 오류 처리, 재시도 등의 기능을 지원함
• Step 간의 순서나 조건에 따라 실행 순서를 제어할 수 있음. 예를 들어, 이전 Step이 성공하면 다음 Step을 실행하고, 실패하면 특정 작업을 수행하도록 설정할 수 있음

    @Bean
    public Step fisrtStep() {
        System.out.println("First Step");
        return new StepBuilder("firstStep", jobRepository)
                .<BeforeEntity, AfterEntity> chunk(10, platformTransactionManager)
                .reader(beforeReader())
                .processor(middleProcessor())
                .writer(afterWriter())
                .build();
    }

4. ItemReader
• 입력 데이터를 받아오는 인터페이스로, 데이터 소스에서 데이터를 읽어오는 역할을 함.
• 예를 들어, 데이터베이스나 파일에서 데이터를 읽어오는 로직을 구현
• 데이터베이스, 파일 시스템, 웹 서비스 등 다양한 소스에서 데이터를 읽어올 수 있음
• 여러 형태의 데이터 소스를 지원하며, 각 소스에 맞는 구체적인 ItemReader 구현체를 제공함
• 데이터를 한 번에 하나씩 읽어오며, 이는 Chunk 기반 처리에서 사용됨

    @Bean
    public RepositoryItemReader<BeforeEntity> beforeReader() {
        return new RepositoryItemReaderBuilder<BeforeEntity>()
                .name("beforeReader")
                .pageSize(10)
                .methodName("findAll")
                .repository(beforeRepository)
                .sorts(Map.of("id", Sort.Direction.ASC))
                .build();
    }

• Cursor, Paging Reader가 있다. 또 Stream도 지원한다(ItemStreamReader).
• 많은 구현체가 있어서 꽤나 다양한 데이터 처리 가능
• JdbcCursorItemReader, JdbcPagingItemReader, HibernateCursorItemReader, HibernatePagingItemReader, JpaPagingItemReader, RepositoryItemReader …
• 하지만 QueryDSL의 구현체는 공식적으로 제공하지 않는 모양이다.
• ‣

5. ItemProcessor
• 읽어온 데이터를 가공하거나 변환하는 역할을 함. 비즈니스 로직이 이 단계에서 처리됨
• 비즈니스 로직을 구현하는 곳으로, 데이터를 필터링하거나 변환하는 작업을 수행함
• ItemReader에서 읽어온 데이터는 ItemProcessor에서 가공되며, 가공된 데이터는 ItemWriter로 전달됨
• 데이터를 필터링하여 제외하거나, 데이터 형식을 변환하거나, 계산 작업을 수행할 수 있음

    @Bean
    public ItemProcessor<BeforeEntity, AfterEntity> middleProcessor() {
        return new ItemProcessor<BeforeEntity, AfterEntity>() {
            @Override
            public AfterEntity process(@NonNull  BeforeEntity item) throws Exception {
                AfterEntity afterEntity = new AfterEntity();
                afterEntity.setUsername(item.getUsername());
                return afterEntity;
            }
        };
    }

6. ItemWriter
• 처리된 데이터를 저장하는 역할을 함. 결과 데이터를 데이터베이스, 파일, 메시지 큐 등에 저장
• 처리된 데이터를 데이터베이스, 파일, 메시지 큐 등 다양한 대상에 저장할 수 있음
• ItemWriter는 데이터를 일괄 처리하거나 하나씩 처리할 수 있으며, Chunk 기반 처리에서 주로 사용됨
• 데이터를 어떻게 저장할지에 대한 로직을 구현하며, 보통 하나의 Step에서 ItemReader와 ItemProcessor와 함께 사용됨

    @Bean
    public RepositoryItemWriter<AfterEntity> afterWriter() {
        return new RepositoryItemWriterBuilder<AfterEntity>()
                .repository(afterRepository)
                .methodName("save")
                .build();
    }

7. 배치 스케줄링
• 구축한 Batch를 스케줄링하는 예시 코드
• 우선 Main 클래스에 @EnableScheduling 어노테이션 추가

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

• 10초마다 스케줄링을 돌리는 예시 코드

   private final JobLauncher jobLauncher;
   private final JobRegistry jobRegistry;

    @Scheduled(cron = "10 * * * * *", zone = "Asia/Seoul")
    public void runFirstJob() throws Exception {

        System.out.println("first schedule start");

        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd-hh-mm-ss");
        String date = dateFormat.format(new Date());

        JobParameters jobParameters = new JobParametersBuilder()
                .addString("date", date)
                .toJobParameters();

        jobLauncher.run(jobRegistry.getJob("firstJob"), jobParameters);
    }

• JobParameter
• 스프링 배치에서 Job 실행 시 외부에서 전달된 값을 나타내는 객체로, 배치 작업의 실행 시점에 필요한 다양한 설정값이나 데이터를 전달하는 데 사용됨
• 특정 JobParameter를 사용해 실행된 Job은 동일한 JobParameter로 다시 실행할 수 없도록 제한됨 → 작업의 일관성 보장