[Spring Batch] 효과적인 ETL 처리를 위한 Spring Batch

ETL 파이프라인 과정에서 사용했던 Spring Batch를 사용하며 경험할 수 있는 장점들을 소개하려고 합니다.
기존에 사용했던 배치 처리 방식에는 linux tab에서 제공해주는 cron job이나 Spring의 @Scheduler 정도만 알고 있었는데, Spring Batch를 쓰면서 배치잡 처리의 더 많은 부분들에 인사이트를 얻을 수 있었습니다. 

---

Spring Batch의 동작 과정

전반적인 핵심 컨셉은 Reader(E) -> Processor(T) -> Writer(L) 의 흐름입니다.

• Reader - 데이터베이스, 파일, 큐에서 다량의 데이터 조회
• Processor - 특정 방법으로 데이터를 가공
• Writer - 데이터를 수정된 양식으로 다시 저장

이 흐름을 실제로 제어해주는 스프링 배치의 단위를 하나씩 살펴봅시다. !~

---

# Job

배치 계층 구조에서 가장 상위 개념으로, 하나의 배치작업 자체를 의미합니다. 여러 Step을 포함하는 컨테이너 역할을 합니다.
ex. 데이터의 앞뒤 빈칸을 제거하고 저장하는 배치 작업: removeWhitespaceDataJob

[JobInstance]

Job을 구분하는 단위입니다. 즉, Job이 일배치로 이루어진다고 할 때 매일마다 생성되는 Job을 JobInstance로 불립니다.

• 처음 시작하는 Job + JobParameter 일 경우 새로운 JobInstance 생성
• 이전과 동일한 Job + JobParameter 으로 실행 할 경우 이미 존재하는 JobInstance 리턴

[JobParameter]

여러개의 JobInstance를 구분하기 위한 용도로 사용되며, 재처리 시 이 개념이 실행됩니다. 특정 JobParameter가 포함된 JobInstance가 성공했다면 재처리 X, 실패했다면 재처리 O

[JobExecution]

JobParameter, Job의 실행 상태, 실행 결과(FAILED, COMPLETED) 등 Job 실행의 전반적인 정보들을 저장해둡니다.

---

# Step

Job 내에 존재하는 단계별 인스턴스로, Job 의 세부 작업을 Task 기반으로 설정하고 명세해 놓은 객체입니다. 트랜잭션은 step 내에서 이뤄집니다. 여러 Job을 하나의 Job 내에서 실행할 수 있는 JobStep, 단일 작업(Tasklet)을 실행하는 TaskletStep, Step들을 하나의 흐름(Flow)으로 묶어 조건부 형태의 흐름제어가 가능한 FlowStep 등 여러 종류가 존재하게 되는데, 구분지어 외울 필요는 없고 상황에 따라 가져와 사용하면 됩니다.

아래에서는 크게 두 가지 Step에 대해 알아보고자 합니다. 

• TaskletStep: 단일 작업 처리 (파일 삭제 등)
• ChunkStep: Reader → Processor → Writer 의 청크 단위 ETL 작업 처리

 

[TaskletStep (Tasklet-Oriented Processing)]

하나의 단일 작업(task)을 구현하고 실행하는 형태로, 단순한 배치 처리에 주로 사용됩니다. Batch의 Step 단계에서 ‘단일한 레코드(row)를 묶어서’ 여러 작업을 처리게 됩니다. 따라서, ‘묶인 레코드를 하나의 트랜잭션으로 처리’하며 실패 하는 경우 롤백을 수행합니다. 실패한 태스크릿만 다시 실행하면 된다는 장점이 있습니다. 아래에서 설명할 chunk가 '하나의 큰 덩어리'로 처리하는 장점이 있다면, tasklet은 '작은 단위로 나누어' 처리할 수 있다는 장점이 있습니다.
ex) 오래된 파일을 삭제하는 tasklet (cleanupTempFilesTasklet)

TaskletStep 동작 방식

[Reader → Processor → Writer (chunk 기반 processing)]

Step 구현 방식 중에서도 reader->processor->writer 방식은 Chunk-oriented processing 에 해당합니다. 따라서 아래와 같이 일정한 단위(chunk size = 100) 기준으로 읽고, 프로세싱하고, 저장하는 흐름으로 구성됩니다.

예를 들어 chunk size가 100이면, 100건씩 ItemReader로 읽기 → ItemProcessor로 정제 → ItemWriter로 저장 까지의 과정이 한 번에 이루어지게 됩니다. 이 단위로 트랜잭션이 묶이기 때문에 중간에 실패하더라도 해당 chunk 단위에서 롤백되며, 전체 Job을 재시작하지 않아도 되는 유연함이 있습니다.
ex) 원시 데이터를 가져오는 Step(reader), 빈칸을 제거하는 Step(processor), 해당 데이터를 저장하는 Step(writer)

---

# Flow

Flow는 여러 Step들을 하나의 흐름 단위로 묶어, 조건 기반 분기 처리를 가능하게 합니다. 예를 들어 Step A가 성공하면 Step B로, 실패하면 Step C로 넘어가게 구성할 수 있으며, 복잡한 배치 흐름을 선언적으로 정의할 수 있습니다.

https://docs.spring.io/spring-batch/reference/step/controlling-flow.html

@Bean
public Job job(JobRepository jobRepository, Step stepA, Step stepB, Step stepC) {
	return new JobBuilder("job", jobRepository)
				.start(stepA)
				.on("*").to(stepB)
				.from(stepA).on("FAILED").to(stepC)
				.end()
				.build();
}

이와 비슷한 개념으로, Decide라는 개념 또한 있는데, Batch Status(FAILED, COMPLETED, STARTING, STARTED 등)가 아닌 본인만의 커스텀한, 동적 exitCode가 필요할 경우 이를 사용할 수 있습니다. 예시는 아래와 같습니다.

class MyDecider : JobExecutionDecider {
    override fun decide(jobExecution: JobExecution, stepExecution: StepExecution?): FlowExecutionStatus {
        return if (Random.nextBoolean()) FlowExecutionStatus("ODD") else FlowExecutionStatus("EVEN")
    }
}
//...
JobBuilder("deciderJob", jobRepository)
    .start(step1())
    .next(myDecider())
    .from(myDecider()).on("ODD").to(stepOdd())
    .from(myDecider()).on("EVEN").to(stepEven())
    .end()

---

간단한 예시 - SubscriptionStatusBatchJob (구독 상태를 변경하는 배치잡)

Spring Batch를 활용하여, 구독 상태를 변경하는 배치잡을 구현해보면 아래와 같습니다. subscriptionExpirationDate가 오늘보다 이전인 값들에 대해 INACTIVE 상태로 변경하는 형태로, 실제 DB를 살펴보면 expiration_date가 오늘보다 이전인 상품들에 대해 INVALID 됨을 볼 수 있습니다.

package com.example.subpay.batch

import com.example.subpay.domain.Subscription
import com.example.subpay.repository.SubscriptionRepository
import org.springframework.batch.core.Job
import org.springframework.batch.core.Step
import org.springframework.batch.core.configuration.annotation.StepScope
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.item.ItemProcessor
import org.springframework.batch.item.ItemReader
import org.springframework.batch.item.ItemWriter
import org.springframework.batch.item.data.builder.RepositoryItemReaderBuilder
import org.springframework.beans.factory.annotation.Value
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.data.domain.Sort
import org.springframework.transaction.PlatformTransactionManager
import java.time.LocalDateTime

@Configuration
class SubscriptionStatusBatchJob(
    private val jobRepository: JobRepository,
    private val transactionManager: PlatformTransactionManager
) {

    @Bean
    fun subscriptionStatusJob(subscriptionStatusStep: Step): Job =
        JobBuilder(JOB_NAME, jobRepository)
            .start(subscriptionStatusStep)
            .build()

    @Bean
    fun subscriptionStatusStep(
        reader: ItemReader<Subscription>,
        processor: ItemProcessor<Subscription, Subscription>,
        writer: ItemWriter<Subscription>
    ): Step =
        StepBuilder(STEP_NAME, jobRepository)
            .chunk<Subscription, Subscription>(CHUNK_SIZE, transactionManager)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .build()

    @Bean
    @StepScope
    fun subscriptionStatusReader(
        subscriptionRepository: SubscriptionRepository
    ): ItemReader<Subscription> =
        RepositoryItemReaderBuilder<Subscription>()
            .name("subscriptionStatusReader")
            .repository(subscriptionRepository)
            .methodName("findAll") // TODO: 영속성이 아닌 Jdbc read 형태로 변경
            .pageSize(CHUNK_SIZE)
            .sorts(mapOf("id" to Sort.Direction.ASC))
            .build()

    @Bean
    @StepScope
    fun subscriptionStatusProcessor(): ItemProcessor<Subscription, Subscription> =
        ItemProcessor { subscription ->
            val now = LocalDateTime.now()
            if (subscription.subscriptionStatus == Subscription.SubscriptionStatus.ACTIVE &&
                subscription.subscriptionExpirationDate.isBefore(now)) {
                subscription.subscriptionStatus = Subscription.SubscriptionStatus.INACTIVE
                subscription
            } else {
                null
            }
        }

    @Bean
    @StepScope
    fun subscriptionStatusWriter(
        subscriptionRepository: SubscriptionRepository,
        @Value("#{jobParameters['dryRun']}") dryRun: String? // jobparameter 활용해 log 먼저 찍어볼 수 있습니다.
    ): ItemWriter<Subscription> {
        val isDryRun = dryRun?.toBooleanStrictOrNull() ?: false

        return ItemWriter { subscriptions ->
            if (isDryRun) {
                println("[DRY RUN] ${subscriptions.size()}개의 구독 상태가 업데이트 될 예정입니다.")
                subscriptions.forEach {
                    println("→ ${it.id} 상태 변경 예정: ${it.subscriptionStatus}")
                }
            } else {
                println("[REAL RUN] ${subscriptions.size()}개의 구독 상태를 저장합니다.")
                subscriptionRepository.saveAll(subscriptions)
            }
        }
    }

    companion object {
        const val JOB_NAME = "subscriptionStatusJob"
        const val STEP_NAME = "subscriptionStatusStep"
        const val CHUNK_SIZE = 100
    }
}

subscription 1개의 구독상태 INVALID로 변경

---

장점

병렬 처리 ⭐️⭐️⭐️⭐️⭐️

Spring 기반 구조이기에, 멀티 쓰레드를 활용해 각 Step을 독립적으로 설계할 수 있어 대규모 데이터 처리 환경에서 유리합니다.

다양한 입출력 옵션(파일, JDBC, NoSQL, JMS 등)을 지원한다. ⭐️⭐️⭐️⭐️⭐️

파일(csv, xml 등), 데이터베이스(JDBC), NoSQL, 메시징 큐(JMS) 등 다양한 입출력 옵션을 지원합니다.

트랜잭션 관리 ⭐️⭐️⭐️⭐️⭐️

각 Step 또는 Chunk 단위에서 트랜잭션이 관리되므로, 일부 데이터가 실패해도 전체 Job이 망가지지 않고 안정적으로 실패 지점만을 복구할 수 있습니다.

Spring Integration과의 통합 ⭐️⭐️⭐️

Spring 기반의 다른 시스템(Spring Cloud, Spring MVC Application)과의 통합이 수월합니다. 

---

REF

https://stackoverflow.com/questions/26929308/advantages-of-spring-batch

https://docs.spring.io/spring-batch/reference/

https://d2.naver.com/helloworld/9879422

https://github.com/jojoldu/spring-batch-in-action, https://jojoldu.tistory.com/328

---

아직은 간단한 배치 잡부터 하나씩 구현하는 중이지만, 필요에 따라 계속 써보며 익혀가 보겠습니다.