Spring Batch 활용

 

다양한 프로젝트를 진행하다 보면, 대용량 데이터 처리를 주기적으로 해야하는 경우가 생긴다. 그럴 경우 흔히 서비스가 돌아가는 메인 서버에서 실행을 하게 되면 서버에 과부하가 걸리지 않을까 하는 생각이 들게 된다. 때문에 이런 경우 Spring Batch 서버를 통해 부하를 분산시켜줄 수 있는데 오늘은 이 Spring Batch가 어떤 것인지와 실제 사용 방식과 예시를 들어서 설명해보겠다. 

---

• Spring Batch란 무엇인가?
  Spring Batch는 대용량 데이터 처리 및 배치 작업을 자동화하는 Spring Framework의 모듈
• 왜 필요한가
  • 데이터 마이그레이션, 정산 처리, 데이터 분석 같은 대규모 배치 작업을 수행하기 위한 안정적이고 확장 가능한 도구.
  • 병렬 처리, 트랜잭션 관리, 재시도 및 장애 복구 등 제공.

쉽게 예시를 통해 설명을 해보면, 

 

내가 당시 참여했던 프로젝트는 문화 예술 관광과 AI를 접목시킨 서비스이었다. 여기서 포인트는 실제 KOPIS 라는 공공 사이트에서 주기적으로(한달) 데이터를 받아와서 저장하고 또 AI로 학습시켜야했다. 문제는 이 한번의 과정이 생각보다 오래걸리고, 서버에 과부하가 걸릴 수 있다는 점이다. 

 

그래서 위의 과정을 하나의 배치(Batch)로 묶어서 Spring Batch 서버에서 돌려주면 훨씬 효율적으로 관리 및 유지 보수가 가능 할 것이라 생각했다.

 

Spring Batch의 주요 구성 요소

• Job
  배치 작업의 상위 구성 요소로, 여러 단계(Step)로 구성
• Step
  하나의 작업 단위로, 읽기(Read), 처리(Process), 쓰기(Write)의 단계를 포함
• ItemReader, ItemProcessor, ItemWriter
  데이터를 읽고(ItemReader), 처리하며(ItemProcessor), 쓰는(ItemWriter) 표준 인터페이스.
• JobRepository와 JobLauncher
  작업 상태와 메타데이터를 관리(JobRepository)하고 작업을 시작(JobLauncher)하는 역할.

사실 Job과 Step만 알면 나머지는 Spring Boot를 많이 사용해본 경험자로서 어렵지 않게 알 수 있었다. 

아래 코드를 통해 두가지를 설명해 보겠다.

 

[Batch Job 설정하기]

  @Bean
    public Job kopisJob(KopisEntityWriter writer) {
        return new JobBuilder("kopisJob", jobRepository)
            .start(firstStep(writer))
            .next(secondStep())
            .next(thirdStep())
            .next(fourthStep())
            .next(fifthStep())
            .build();
    }

 

 

Job의 이름은 KopisJob으로 되어 있고 해당 Job에서는 총 5가지 Step이 순서대로 작동한다고 생각하면 된다. 

 

그리고 Step 같은 경우에도 마찬가지로 아래와 같이 쉽게 정의할 수 있다.

 

[Batch Step 정의하기]

// 1. Kopis에서 해당 기간에 대한 공연 정보 가져 오기
    @Bean
    public Step firstStep(KopisEntityWriter writer) {
        return new StepBuilder("firstStep", jobRepository)
            .<Concert, Concert>chunk(10, platformTransactionManager)
            .reader(kopisApiReader)
            .writer(writer)
            .build();
    }

 

나는 첫번째 Step으로 방금말한 Kopis라는 공공사이트에서 API를 호출 함으로서 주기적으로 공연 정보를 담아올 수 있도록 했다. 

자세한 로직은 kopisApiReader에 정의되어 있으나 여기선 중요하지 않으니 생략하기로 하겠다. 

 

그래서 나는 내 Job Step을 다음과 같이 나눴다.

[Kopis Job]

1. Kopis에서 공연데이터 API 호출

2. Kopis에서 받아온 공연데이터 상세정보 API 호출

3. 공연 포스터 이미지 OCR 및 AI 학습 작업 

4. 추출된 정보 CSV 파일 저장

5. 공연 정보 AI 요약 DB Mapping

 

전체코드는 다음과 같다

@Bean
    public Job kopisJob(KopisEntityWriter writer) {
        return new JobBuilder("kopisJob", jobRepository)
            .start(firstStep(writer))
            .next(secondStep())
            .next(thirdStep())
            .next(fourthStep())
            .next(fifthStep())
            .build();
    }

    // 1. Kopis에서 해당 기간에 대한 공연 정보 가져 오기
    @Bean
    public Step firstStep(KopisEntityWriter writer) {
        return new StepBuilder("firstStep", jobRepository)
            .<Concert, Concert>chunk(10, platformTransactionManager)
            .reader(kopisApiReader)
            .writer(writer)
            .build();
    }

    // 2. Step1 에서 가져온 공연에 대해 상세 내역 가져 오기
    @Bean
    public Step secondStep() {
        return new StepBuilder("secondStep", jobRepository)
            .tasklet(kopisDetailApiReader, platformTransactionManager)
            .build();
    }

    // 3. 공연 포스터 이미지를 통해서 Flask 서버로 부터 카테고리 추출
    @Bean
    public Step thirdStep() {
        return new StepBuilder("thirdStep", jobRepository)
            .tasklet(concertCategoryExtractorTasklet(), platformTransactionManager)
            .build();
    }

    // 4. 추출한 카테고리 값을 CSV 파일에 저장
    @Bean
    public Step fourthStep() {
        return new StepBuilder("fourthStep", jobRepository)
            .tasklet((StepContribution contribution, ChunkContext chunkContext) -> {

                String folderPath = "datasets";
                String fileName = "concerts.csv";

                String localFilePath = s3Service.downloadCsvFile(folderPath, fileName);
                if (!isFileExists(localFilePath)) {
                    return RepeatStatus.FINISHED;
                }

                StringJoiner csvContent = readExistingCsvContent(localFilePath);
                appendNewConcertData(csvContent);

                saveAndUploadCsvFile(csvContent, folderPath, fileName);

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

    // 5. 클라코 큐레이션: 공연 정보 요약
    @Bean
    public Step fifthStep() {
        return new StepBuilder("fifthStep", jobRepository)
            .tasklet(concertSummaryExtractorTasklet(), platformTransactionManager)
            .build();
    }

 

그래서 여기까지가 Batch의 Job과 Step을 정의하는 과정인데, 중요한 것은 이 Batch를 어떤 주기로 돌려줄것인가이다. 

이 내용은 Schedule이라는 Java Class에서 따로 정의해주면 되는데 코드는 아래와 같다.

 

[Spring Scheduler 설정하기]

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

        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("kopisJob"), jobParameters);
    }

 

@Scheduled 가 의미하는 것은 다음과 같다. 현재는 10초마다 돌려주는 것으로 되어 있는데, 실제 운영환경에서는 아래 주석부분에서 처럼 한달에 한번씩 돌려줄 경우이다. 결론적으로 "cron에 맞춰 jobLauncher를 돌려주겠다" 라는 표현이 되겠다. 

---

그래서 Spring Batch의 장점과 특징에 대해서 정리를 해보면 다음과 같다.

Spring Batch의 주요 특징

• 재시도 및 장애 복구
  실패한 작업을 재시작하거나 특정 단계부터 다시 실행할 수 있음.
  예: 데이터베이스 트랜잭션 롤백.
• 병렬 처리
  멀티스레드와 파티셔닝을 통해 대규모 데이터를 효율적으로 처리.
• 확장 가능성
  여러 데이터 소스 및 대상과 통합 가능 (예: 파일, 데이터베이스, API).

확실히 대용량 데이터를 주기적으로 핸들링 해줘야하는 이번 프로젝트에 확실히 유용하게 쓸 수 있었다. 

이번 Batch 의 Scheduler에서는 cron을 이용하여 JobLaucher를 돌렸는데, 우리가 흔히 생각하는 Controller에서도 JobLaucher를 돌릴 수 있다고 하니, 추후 사용해볼 예정이다.