📌Spring Batch?

Spring Batch는 대용량 데이터 처리 애플리케이션을 구축하기 위한 프레임워크로, 주로 배치 처리에 사용됩니다. 배치 처리란 대규모 데이터를 일괄적으로 처리하는 작업으로, 보통 일정한 시간에 실행되거나 특정 조건이 충족될 때 자동으로 수행됩니다. 예를 들어, 야간에 대규모 데이터 업데이트, 보고서 생성, 파일 처리 등이 이에 해당합니다.

📌사용 이유

📍대용량 데이터 처리

Spring Batch는 대규모 데이터를 효율적으로 처리할 수 있는 기능을 제공합니다. 예를 들어, 수백만 건의 데이터베이스 레코드, 파일, 메시지 큐를 처리할 때 사용됩니다.

📍재시작 가능성

Spring Batch는 작업이 실패했을 때 중단된 지점부터 다시 시작할 수 있는 기능을 제공합니다. 이를 통해 처리 중 장애가 발생해도 안정적으로 작업을 완료할 수 있습니다.

📍트랜잭션 관리

Spring Batch는 트랜잭션을 관리하여 배치 작업의 일관성을 보장합니다. 특정 단계에서 문제가 발생하면, 해당 단계에서 수행된 모든 작업을 롤백할 수 있습니다.

📍청크 기반 처리

데이터를 청크(작은 단위)로 나누어 처리합니다. 청크 단위로 트랜잭션을 관리하여 메모리 효율성을 높이고, 처리 중단 시 복구가 용이하게 합니다.

📍확장성

Spring Batch는 멀티스레드, 멀티 프로세싱, 클러스터링 등을 지원하여 확장성을 높입니다. 이를 통해 대규모 배치 작업을 더 빠르고 효율적으로 처리할 수 있습니다.

📍통합 및 모듈화

Spring Batch는 기존 Spring 애플리케이션과 쉽게 통합할 수 있습니다. 모듈화된 구조를 가지고 있어, 특정 기능만을 선택적으로 사용할 수 있습니다.

📌주요 개념

📍Job

배치 처리 작업의 단위로, 여러 개의 스텝으로 구성됩니다.

📍Step

Job의 하위 작업 단위로, 실제 데이터 처리 로직이 구현됩니다. 일반적으로 Reader, Processor, Writer로 구성됩니다.

📍ItemReader

데이터를 읽어오는 역할을 합니다. 예를 들어, 파일, 데이터베이스, 메시지 큐 등에서 데이터를 읽어옵니다.

📍ItemProcessor

읽어온 데이터를 가공하거나 변환하는 역할을 합니다.

📍ItemWriter

처리된 데이터를 출력하는 역할을 합니다. 예를 들어, 데이터베이스에 저장하거나 파일로 기록할 수 있습니다.

📍JobRepository

Job의 메타데이터를 저장하고 관리하는 역할을 합니다.

📍JobLauncher

Job을 실행하는 데 사용됩니다.

📌Spring Batch의 구조와 동작

Spring Batch는 크게 Job과 Step으로 구성되며, 각각의 Job은 여러 Step으로 나뉩니다. Step은 순차적으로 실행되며, 각 Step은 Reader, Processor, Writer로 이루어져 있습니다.

📍Job

배치 프로세스의 논리적 단위로, 여러 개의 Step을 포함합니다.

📍Step

Job의 구성 요소로, 각 Step은 특정 작업을 수행합니다.

📍Reader

데이터를 읽어옵니다.

📍Processor

데이터를 변환하거나 가공합니다.

📍Writer

데이터를 저장하거나 출력합니다.

📌Example

그럼 예시 코드를 보면서 Spring Batch 의 사용 방법에 대해 알아보겠습니다.
Spec Springboot 3.3.3, Gradle Project

Spring Batch를 사용하여 Database 데이터를 삭제하고 수집한 새로운 데이터로 삽입하는 배치작업 예제입니다.

Add Dependency

spring-boot-stater-batch 의존성을 추가합니다. 편의를 위한 Lombok 과 Database 연결을 위해 Mariadb client도 추가합니다.

//Spring Batch
implementation 'org.springframework.boot:spring-boot-starter-batch'

//Lombok
compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.projectlombok:lombok'

// MariaDB
runtimeOnly 'org.mariadb.jdbc:mariadb-java-client'

Add Application options

application.yml or properties 파일에 배치 관련 옵션을 추가합니다.

spring:
   batch:
    job:
      enabled: true
    jdbc:
      initialize-schema: always

batch.job.enabled: true
이 설정은 Spring Batch 작업 기능을 활성화하는 옵션입니다. true로 설정하면 애플리케이션이 시작될 때 자동으로 배치 작업이 실행됩니다. 만약 false로 설정하면 애플리케이션이 시작되더라도 배치 작업이 자동으로 실행되지 않습니다.

jdbc.initialize-schema: always
이 설정은 데이터베이스 스키마를 어떻게 초기화할지를 지정하는 옵션입니다. always로 설정하면 애플리케이션이 실행될 때마다 데이터베이스 스키마가 항상 초기화됩니다. 여기에는 테이블 생성, 삭제, 수정 등이 포함될 수 있습니다.

📍BatchConfig

Batch 처리를 위한 BatchConfig class 를 생성합니다.
config.batch.BatchConfig

@Configuration
@EnableBatchProcessing
@RequiredArgsConstructor
@Slf4j
public class BatchConfig {
//이어서..

Spring Batch 를 활성화 하기 위해선 @EnableBatchProcessing 을 추가해야 합니다. @Configuration 을 제외한 나머지 Annotation 들은 lombok의 Annotation 입니다.

@Configuration
@EnableBatchProcessing
@RequiredArgsConstructor
@Slf4j
public class BatchConfig {
    private final JdbcTemplate jdbcTemplate;
    private final DataCollectionService dataCollectionService;

JdbcTemplate 와 특정 API 로부터 데이터를 수집할 DataCollectionService 를 DI (Dependency Injection)합니다. JdbcTemplate 는 Spring 에서 제공하는 JDBC API로, Database 와의 상호작용을 단순화해줍니다. Databse table 에 데이터 Delete, Insert 시 활용합니다.
DataCollectionService는 특정 API 로 부터 데이터를 수집하는 역할을 합니다. 자세한 설명은 생략합니다.

🎈Job 등록

@Bean
public Job dataUpdateJob(JobRepository jobRepository
					   , PlatformTransactionManager transactionManager) {
    return new JobBuilder("wifiUpdateJob", jobRepository)
            .start(deleteStep(jobRepository, transactionManager))
            .next(insertStep(jobRepository, transactionManager))
            .build();
}

Batch 의 작업단위인 Job 을 등록 합니다. JobBuilder 로 등록하며 Delete와 Insert 로직을 Step으로 구분하여 등록합니다. start로 등록한 deleteStep 실행 후에 next 로 등록한 insertStep 이 동작합니다. JobRepository는 배치 메타데이터를 저장하는 장소이고 PlatformTransactionManager 트랜잭션 처리를 담당합니다.

🎈Steps

@Bean
public Step deleteStep(JobRepository jobRepository
		, PlatformTransactionManager transactionManager) {
    return new StepBuilder("deleteStep", jobRepository)
            .tasklet((contribution, chunkContext) -> {
                jdbcTemplate.update("DELETE FROM TABLE");
                return RepeatStatus.FINISHED;
            }, transactionManager)
            .build();
}

@Bean
public Step insertStep(JobRepository jobRepository
		, PlatformTransactionManager transactionManager) {
    return new StepBuilder("insertStep", jobRepository)
            .<WifiResponse, WifiResponse>chunk(100, transactionManager)
            .reader(itemReader())
            .writer(itemWriter())
            .listener(new StepExecutionListener() {
                @Override
                public void beforeStep(StepExecution stepExecution) {
                    log.info("Before Step: insertStep");
                }

                @Override
                public ExitStatus afterStep(StepExecution stepExecution) {
                    log.info("After Step: insertStep, Status: {}"
                    	, stepExecution.getStatus());
                    return stepExecution.getExitStatus();
                }
            })
            .build();
}

Database table에 데이터 delete, insert 하는 step 입니다. deleteStep에 있는 tasklet 은 step 내에서 실행되는 단일 task 수행 Interface 입니다. delete 만 요청하는 간단한 작업이기 때문에 tasklet을 활용합니다.
insertStep은 chunk를 사용하여 100건의 데이터씩 처리를 합니다. reader는 데이터를 읽어오고, writer는 데이터를 데이터베이스에 insert 하는 역할을 합니다. listener를 등록하여 step이 실행되기 전후에 로그를 출력합니다.

🎈ItemReader, ItemWriter

@Bean
@StepScope
public ListItemReader<DataResponse> itemReader() {
    List<DataResponse> items = new ArrayList<>();
    try {
        items = this.dataCollectionService.getData();
        log.info("Inserted {} records into TABLE table.", items.size());
    } catch (IOException e) {
        log.error("Data collection failed.", e);
    }
    return new ListItemReader<>(items);
}

@Bean
public ItemWriter<DataResponse> itemWriter() {
    return items -> {
        try {
            for (DataResponse item : items) {
                jdbcTemplate.update("INSERT INTO TABLE " + 
                	"(COL_01, COL_02, COL_03, COL_04, COL_05, COL_06, COL_07) " +
                	"VALUES (?, ?, ?, ?, ?, 'Y', now())",
                    item.col1(), item.col2(), item.col3()
                  , item.col4(), item.col5());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    };
}

ListItemReader의 데이터는 한번 읽으면 다시 사용할 수 없기 때문에 최초 1회만 itemReader 가 동작하게 됩니다. 매번 동작하게 하기 위해선 @StepScope 를 추가해서 itemReader가 스텝이 실행될 때마다 새로운 인스턴스를 생성하도록 해야 합니다.

itemReader 를 통해서 데이터를 수집합니다. dataCollectionService의 getData method를 통해 List<DataResponse> 를 받고, 받은 데이터를 ListItemReader로 감싸서 반환합니다.

itemReader를 통해 읽어온 데이터를 itemWriter에서 Database에 insert 합니다. jdbcTemplate.update() 를 사용해 직접 SQL을 실행합니다.

🎈TransactionManager

@Bean
public PlatformTransactionManager transactionManager(DataSource dataSource) {
    return new DataSourceTransactionManager(dataSource);
}

PlatformTransactionManager는 트랜잭션 관리를 담당합니다. 여기서는 DataSourceTransactionManager를 사용하여 JDBC 기반의 트랜잭션을 관리합니다.

이제 Spring Batch 의 Job을 위한 구현은 완료되었습니다. 이제 주기적으로 해당 Job을 실행하기 위해 Scheduler 를 등록하고 jobLauncher 를 활용하여 해당 Job을 실행하는 코드를 작성하겠습니다.

📌CollectionScheduler

scheduler.CollectionScheduler

@Component
@RequiredArgsConstructor
@EnableScheduling
public class CollectionScheduler {

    private final JobLauncher jobLauncher;
    private final Job dataUpdateJob;

    @Scheduled(cron = "5 0 0 * * ?")
    public void runJob() throws JobExecutionException {
        this.jobLauncher.run(this.dataUpdateJob, new JobParametersBuilder()
                //Batch 실행 시 Parameter 가 같으면 같은 동작으로 인식. 매번 새로운 파라메터 전달
                .addLong("time", System.currentTimeMillis())
                .toJobParameters());
    }

    /**
     * Application 실행 시 최호 1회 Spring Batch 실행
     */
    @Bean
    public ApplicationRunner runJobOnStartup() {
        return args -> {
            this.jobLauncher.run(this.dataUpdateJob, new JobParametersBuilder()
                    .addLong("time", System.currentTimeMillis())
                    .toJobParameters());
        };
    }
}

📍@EnableScheduling, @Scheduled

Scheduler 가 실행되게 하기 위해 @EnableScheduling 를 class level에 추가하고 JobLauncher 와 위에서 등록한 dataUpdateJob을 의존성 주입합니다.
@Schduled 를 활용하여 매일 00:05 분에 실행할 method를 등록합니다. method 내에서는 jobLauncher 를 활용하여 job 을 실행합니다. 주석과 같이 run method 실행 시 parameter를 전달하는 이유는 Spring Batch 에서 동일한 Job을 여러 번 실행 할 수 있게 하기 위함입니다. Spring Batch는 기본적으로 동일한 Job의 중복실행을 방지합니다. Job 은 고유한 Job Instance로 관리되는데, 이 Job Instance는 Job 이름과 Job Parameters의 조합에 따라 생성됩니다. 만약 Parameter 가 동일하다면, 이미 실행된 Job Instance로 판단하여 재실행하지 않습니다. 그렇기 때문에 Parameter로 System.currentTimeMillis() 를 전달하여 Job을 여러 번 실행될 수 있도록 보장합니다.

📍ApplicationRunner

Application 실행 시 최초 1회 Job을 동작시키기 위해 ApplicationRunner를 등록합니다.

이제 Spring Batch 가 실행되면 아래와 같은 로그를 확인하실 수 있습니다.

Job: [SimpleJob: [name=dataUpdateJob]] launched with the following parameters...
Executing step: [deleteStep]
Step: [deleteStep] executed in 7ms
Executing step: [insertStep]
Before Step: insertStep
Inserted 567 records into TABLE table.
After Step: insertStep, Status: COMPLETED
Step: [insertStep] executed in 648ms
ob: [SimpleJob: [name=wifiUpdateJob]] completed with the following parameters...

📌Conclusion

Spring Batch는 대량의 데이터 처리 작업을 효율적으로 관리하고 자동화할 수 있도록 도와주는 강력한 도구입니다. Job과 Step의 구조로 복잡한 비즈니스 로직을 구성하고, 배치 처리의 상태를 관리하여 재시작이 가능하도록 설계되었습니다. 특히, Job Parameters를 통해 동일한 작업을 유연하게 반복 실행할 수 있으며, 이를 통해 데이터 처리의 안정성과 확장성을 확보할 수 있습니다. 다양한 스케줄링과 트랜잭션 관리 기능을 결합해 배치 작업의 복잡성을 효과적으로 해결할 수 있는 솔루션입니다.

📚참고

📕JDBCTemplate

JdbcTemplate은 스프링 프레임워크(Spring Framework)에서 제공하는 데이터베이스 연동을 위한 유틸리티 클래스로, JDBC(Java Database Connectivity) API를 더 간편하고 안전하게 사용할 수 있도록 도와줍니다. JDBC를 이용한 데이터베이스 연동 과정에서 자주 발생하는 반복적인 코드와 오류를 줄이고, 예외 처리를 일관되게 할 수 있는 장점이 있습니다.

📖주요 특징 및 장점

리소스 관리 자동화
JdbcTemplate은 데이터베이스 연결, 명령 실행, 결과 처리 후 리소스를 자동으로 정리해줍니다. 이를 통해 개발자가 직접 Connection, Statement, ResultSet을 관리하지 않아도 됩니다. 따라서 리소스 누수 가능성이 줄어듭니다.

반복적인 코드 제거
JDBC를 사용할 때마다 발생하는 반복적인 코드, 예를 들어 SQL 실행, 예외 처리, 결과 세트 반복 처리 등을 JdbcTemplate이 대신 처리해줍니다. 이렇게 하면 코드가 더 간결해지고 유지보수가 쉬워집니다.

일관된 예외 처리
JdbcTemplate은 SQL 예외를 스프링의 데이터 액세스 예외 계층으로 변환합니다. 이를 통해 개발자는 데이터베이스 벤더에 독립적인 예외 처리를 할 수 있습니다.

다양한 데이터베이스 작업 지원
JdbcTemplate은 단순 조회, 삽입, 업데이트, 삭제 작업 외에도 배치 업데이트, 트랜잭션 관리 등 다양한 데이터베이스 작업을 지원합니다.

📙Crontab

크론탭은 유닉스(Unix) 계열 운영체제에서 주기적인 작업을 자동으로 실행하기 위해 사용하는 스케줄링 도구입니다. "크론(Cron)"은 시간 기반 작업 스케줄러이며, "탭(Tabl)"은 테이블을 의미합니다. 즉, 크론탭은 스케줄링을 위한 작업 목록 파일을 관리하는 역할을 합니다. 기본 cron 은 5자리를 사용하지만 Quartz Scheduler 같은 확장된 스케쥴러는 6자리를 지원합니다.

📖크론탭 표현식 구조

* * * * * *
초 분 시 일 월 요일
크론탭 표현식은 6개의 필드로 구성되어 있으며, 각 필드는 일정한 시간 단위를 나타냅니다. 각 필드의 의미는 다음과 같습니다
초 (Second): 0-59
분 (Minute): 0-59
시간 (Hour): 0-23
일 (Day of Month): 1-31
월 (Month): 1-12
요일 (Day of Week): 0-7 (0 또는 7은 일요일)

? 는 주로 월, 요일 위치에 사용되며 아무 값도 지정하지 않음, 해당 필드에 상관없음 을 의미합니다.

📖Example

*/5 * * * * * -> 매 5초마다 실행
0 * * * * * -> 매 1분마다 실행
10 * * * * * -> 매 10초마다 0분에 실행
0 0 0 * * * -> 매일 자정(00:00:00) 에 실행
10 30 9 * * 1 -> 매주 월요일 오전 9시 30분 10초에 실행
0 0 12 1 * * -> 매월 1일 12시 0분 0초에 실행 (12:00:00)
30 15 6 25 12 * -> 매년 12월 25일 오전 6시 15분 30초에 시행
0 15 10 * * ? -> 매일 오전 10시 15분에 실행 (요일 상관 없음)
0 30 14 ? * 1 -> 매주 월요일 오후 2시 30분에 실행 (일 상관 없음)