Springbatch에서 메타테이블 없이 실행(Springbatch 5)

외부로부터 대량의 데이터를 가져와 주기적으로 가공 및 저장하는 기능 개발이 필요했습니다.

이에 Springbatch를 사용해서 배치 프로그램을 작성 해보기로 했습니다.

 

다만, 문제가 있었는데요.

자주 실행 되는 만큼 메타 테이블에 데이터가 너무 많이 쌓인다는 것이었어요.

메타 테이블이란 Springbatch 프레임워크에서 작업의 상태를 추적하고 관리하기 위해 메타데이터를 저장하는 장소입니다.
BATCH_JOB_INSTANCE, BATCH_JOB_EXECUTION, BATCH_STEP_EXECUTION ... 등으로 구성되어 있습니다.
기본적으로 Job을 실행하면 해당 테이블에 관련 데이터들이 자동으로 생성 및 저장됩니다.

 

메타테이블을 사용한다면 거기에 따른 장점이 있겠지만

현시점에서 중요도는 떨어져서 메타 데이터에 대한 저장없이 배치 프로그램을 작성하는 방법을 택했습니다.

 

시작하기전에 애플리케이션 구동 환경은 대략 아래와 같습니다.

• Springboot 3.1.4
• Springbatch 5
• JDK 17
• Kotlin 1.8
• Mysql 8.0.28

---

1. 의존성 & Application.yml

implementation("org.springframework.boot:spring-boot-starter-batch")
implementation("org.springframework.boot:spring-boot-starter-data-jpa")
implementation("mysql:mysql-connector-java:8.0.28")

위 의존성을 build.gradle.kts에 추가해줬습니다.

 

그리고 application.yml 은 아래와 같이 작성했습니다.

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/testdb
    username: 유저명
    password: 패스워드
    driver-class-name: com.mysql.cj.jdbc.Driver
  jpa:
    database: mysql
    database-platform: org.hibernate.dialect.MySQLDialect
    show-sql: true
    hibernate:
      ddl-auto: none

 

2. Main, 프로그램 진입점

배치프로그램 진입 지점 작성하는 방법은 다양할 것입니다.

저는 program argument로 받은 변수를 이용해서 거기에 맞는 Job을 실행하고자

아래와 같이 작성했습니다.

@SpringBootApplication
@EnableBatchProcessing
class BatchApplication

// program argument로 부터 job name 추출
fun getJobName(args: Array<String>): String? {
    for (arg in args) {
        if (arg.startsWith("--job.name")) {
            return arg.split("=")[1]
        }
    }
    return null
}

// 현재 시각으로 job parameter 추출
fun getJobParameter(): JobParameters {
    val simpleDateFormat: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS")
    val now = LocalDateTime.now().atZone(ZoneId.of("Asia/Seoul")).format(simpleDateFormat)
    return JobParametersBuilder().addString("execute.now", now).toJobParameters()
}

fun main(args: Array<String>) {
    val app = SpringApplicationBuilder(BatchApplication::class.java).web(WebApplicationType.NONE)
    val context = app.run(*args)

    try {
    	// jobLauncher로 입력받은 job이름에 해당하는 Bean을 찾아 실행
        val jobLauncher = context.getBean("jobLauncher") as JobLauncher
        val jobName: String = getJobName(args) ?: throw Exception("invalid job name")
        val job: Job = context.getBean(jobName) as Job
        val execution = jobLauncher.run(job, getJobParameter())
        
        println("success job: ${execution.status}")
    } catch (e: Exception) {
        println("failed job")
    }
}

 

이후 빌드 하게 되면

아래와 같이 jar 파일을 실행하여 Job을 실행할 수 있습니다.

java -jar ./build/libs/application-0.0.1.jar --job.name=simpleJob

 

하지만 아직 simpleJob이 없어서 invalid job name 에러가 뜨겠죠.

 

3. Job 생성

테스트를 위해 아래와 같은 Job 및 Step을 추가해줍니다.

@Configuration
class JobConfig {

    @Bean
    fun simpleJob(jobRepository: JobRepository, simpleStep: Step): Job {
        return JobBuilder("simpleJob", jobRepository)
            .start(simpleStep1)
            .build()
    }

    @Bean
    fun simpleStep(
        jobRepository: JobRepository,
        testTasklet: Tasklet,
        platformTransactionManager: PlatformTransactionManager
    ): Step {
        return StepBuilder("simpleStep", jobRepository)
            .tasklet(testTasklet, platformTransactionManager).build()
    }

    @Bean
    fun testTasklet(): Tasklet {
        return (Tasklet { contribution, chunkContext ->
            println("tasklet")
            RepeatStatus.FINISHED
        })
    }
}

별다른 기능은 없습니다.

단순히 메타데이터 저장 없이 job을 실행해보기 위한 것이니까요.

 

4. JobRepository 관련 에러

실행 해봅시다.

아래와 같은 에러가 뜹니다.

java.lang.IllegalStateException: Failed to execute ApplicationRunner
	at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:768) ~[spring-boot-3.1.4.jar:3.1.4]
	at org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:755) ~[spring-boot-3.1.4.jar:3.1.4]
	at org.springframework.boot.SpringApplication.run(SpringApplication.java:322) ~[spring-boot-3.1.4.jar:3.1.4]
	at org.springframework.boot.builder.SpringApplicationBuilder.run(SpringApplicationBuilder.java:150) ~[spring-boot-3.1.4.jar:3.1.4]
	at me.multimoduleexam.modulebatch.BatchApplicationKt.main(BatchApplication.kt:39) ~[main/:na]
Caused by: org.springframework.jdbc.BadSqlGrammarException: PreparedStatementCallback; bad SQL grammar [SELECT JOB_INSTANCE_ID, JOB_NAME
FROM BATCH_JOB_INSTANCE
WHERE JOB_NAME = ?
...(생략)

메타데이터를 저장할 테이블이 존재하지 않아 발생한 에러네요.

 

하지만 메타 테이블을 사용하지 않을 것이기 때문에 

추가적으로 코드를 작성해줘야합니다.

 

5. 어떻게?

저는 h2를 사용해 볼 것입니다.

즉, 인메모리DB를 사용해서 메타데이터를 저장하고 프로그램 종료와 함께 유실(?)되도록 말입니다.

그러기 위해서는 배치 프로그램 실행 시 두 개의 Datasource 빈이 생성되어야 됩니다.

• 메인 Datasource: 실제 배치 프로그램 작업으로 인한 결과물이 저장. 여기서는 Mysql
• 서브 Datasource: 메타데이터를 임시적으로 저장. H2

 

6. 의존성 추가

h2 의존성을 추가해줍시다.

implementation("com.h2database:h2:2.2.224")

 

 

7. 설정 추가

아래와 같이 핵심이 될 코드를 추가해줍니다.

@Order(-1)
@Configuration
class BatchConfig(
    @Value("\${spring.datasource.url:}") val dbUrl: String,
    @Value("\${spring.datasource.username:}") val dbUsername: String,
    @Value("\${spring.datasource.password:}") val dbPassword: String
) {
	// 메인 데이터소스
    @Primary
    @Bean
    fun dataSource(): DataSource {
        val hikariConfig = HikariConfig()
        hikariConfig.driverClassName = "com.mysql.cj.jdbc.Driver"
        hikariConfig.jdbcUrl = dbUrl
        hikariConfig.username = dbUsername
        hikariConfig.password = dbPassword
        return HikariDataSource(hikariConfig)
    }
	
    // 서브 데이터소스
    @Bean(name = ["subDatasource"])
    fun subDatasource(): DataSource {
        val embeddedDatabaseBuilder = EmbeddedDatabaseBuilder()
        return embeddedDatabaseBuilder.setType(EmbeddedDatabaseType.H2)
            .addScript("/org/springframework/batch/core/schema-h2.sql")
            .build()
    }

    @Bean
    fun transactionManager(): PlatformTransactionManager {
        return JpaTransactionManager()
    }

	// 서브데이터 소스를 사용하도록 설정
    @Bean
    fun jobRepository(): JobRepository {
        val factory = JobRepositoryFactoryBean()
        factory.setDataSource(subDatasource())
        factory.transactionManager = transactionManager()
        factory.afterPropertiesSet()
        return factory.getObject()
    }

    @Bean
    fun jobLauncher(): JobLauncher {
        val jobLauncher = TaskExecutorJobLauncher()
        jobLauncher.setJobRepository(jobRepository())
        jobLauncher.afterPropertiesSet()
        return jobLauncher
    }
}

메인 데이터소스가 될 빈에는 반드시 @Primary 어노테이션을 붙여줘야 합니다.

그렇지 않으면 jobRepository가 계속해서 메인 Datasource를 통해 연결을 시도할 것입니다.

 

서브 데이터소스는 Bean name을 다르게 지정해줍시다.

저는 subDatasource로 해줬습니다.

 

그리고 jobRepository 빈을 명시해줍니다.

반드시 subDatasource로 등록해줍시다.

 

 

8. 실행

2024-03-19T23:07:47.393+09:00  INFO 9887 --- [           main] m.m.modulebatch.BatchApplicationKt       : Started BatchApplicationKt in 1.623 seconds (process running for 1.867)
2024-03-19T23:07:47.419+09:00  INFO 9887 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=simpleJob]] launched with the following parameters: [{'execute.now':'{value=2024-03-19 23:07:47.394, type=class java.lang.String, identifying=true}'}]
2024-03-19T23:07:47.437+09:00  INFO 9887 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [simpleStep]
tasklet
2024-03-19T23:07:47.447+09:00  INFO 9887 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [simpleStep] executed in 10ms
2024-03-19T23:07:47.455+09:00  INFO 9887 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=simpleJob]] completed with the following parameters: [{'execute.now':'{value=2024-03-19 23:07:47.394, type=class java.lang.String, identifying=true}'}] and the following status: [COMPLETED] in 28ms
success job: COMPLETED

정상적으로 잘 실행되는 것을 확인 할 수 있습니다.