Spring Batch의 Job 알아보기

Job의 특성

Job은 독립적으로 실행할 수 있는 스텝의 목록이다. 아래의 특성을 바탕으로 조금 더 이해해 보자.

1. 유일성: Job은 Bean 객체 구성 방식과 동일하게, 한번 구성된 Job은 여러 차례 재사용될 수 있다.
2. 순서를 가진 Step의 목록:  여러 개의 Step으로 구성되며, Step은 정해진 순서에 따라 실행된다.
3. 실행의 단위: 실행 상태에서 외부 이벤트를 기다리지 않으며, 외부 이벤트가 발생한 경우 실행된다.
4. 독립적: Job은 외부 의존적이어서 생길 수 있는 경우의 수에 대해서 제어해야 한다.

 

Job 생명주기

Job을 실행시키는 시작점은 Job Runner이다. Spring Batch에서는 기본적으로 아래 두 개의 JobRunner를 제공한다.

• CommandLineJobRunner
  • 스크립트 또는 CLI에서 직접 Job을 실행하는 경우 사용
  • Spring Application을 Bootstrap하고, 전달받은 파라미터를 기준으로 Job 실행
• JobRegistryBackgroundJobRunner
  • Spring Application이 Bootstrap된 상태에서 스케줄러를 사용해서 Job을 실행하는 경우 사용
  • Bootstrap되는 시점에 JobRegistry를 생성하는 역할을 수행

참고: https://docs.spring.io/spring-batch/reference/job/running.html

Running a Job :: Spring Batch

 

위 공식 문서에서는 Job을 실행시키는 조건으로 필요한 것을 두 가지로 정리하는데, 그것은 Job을 실행시키는 JobLauncher와 실행시킬 Job이다. 그리고 CommandLineJobRunner의 동작 순서는 아래와 같이 설명한다.

Spring Batch에서 표준으로 제공하는 실행 진입점은 JobRunner가 아니라 JobLauncher이다. 따라서 JobRunner를 Spring Batch에서 제공하긴하지만, 표준 실행 진입점으로 제공하는 것은 아니다.

예를 들어 아래와 같이 Spring MVC에서도 JobLauncher를 활용하여 Job을 실행시킬 수 있다.

@Controller
public class JobLauncherController {

    @Autowired
    JobLauncher jobLauncher;

    @Autowired
    Job job;

    @RequestMapping("/jobLauncher.html")
    public void handle() throws Exception{
        jobLauncher.run(job, new JobParameters());
    }
}

 

이처럼 Job을 실행하게 된다면 JobInstance가 생성된다. JobInstance는 Job 이름과 Job 파라미터를 기준으로 표현하는 Job의 논리적인 실행이다.

또한 JobInstance는 각 실행마다 JobExecution을 가지며 성공적으로 완료된 JobExecution이 있다면 완료된 것으로 간주한다.

 

JobRepository 설정

Job을 구성하기에 앞서 JobRepository를 in-memory DB가 아닌 RDBMS로 설정하는 방법을 알아보자.

spring:
  datasource:
    driverClassName: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/spring_batch_example
    username: root
    password: root
  batch:
    initialize-schema: always

application.yaml 파일을 위와 같이 설정하여 Spring Application이 Datasource 빈을 초기화하도록 한다.

그리고 잡을 실행하게 된다면 적절한 스키마에 Job과 관련된 정보들이 갱신될 것이다.

Spring Batch의 JobRepository가 활용하는 테이블 목록

Job 관련 데이터 적재 현황

 

JobParameters 전달하기

Job에 적절한 파라미터를 전달하기 위한 여러 가지 방법이 존재한다. 우선 JobLauncerCommandLineRunner 기준으로 이해해보자.

java -jar batch_example.jar fileName=target_data.csv

위와 같이 명령어에 key=value 형식으로 파라미터를 전달할 수 있다.

사용자가 JobRunner에 전달한 파라미터는 JobParameters 인스턴스로 변환되어 Job에 전달된다.

 

Job에서는 전달받은 JobParameters에서 적절하게 JobParameter 인스턴스를 꺼내어 사용한다.

(JobParameters는 Map<String, JobParameter>의 Wrapper Class이다)

 

JobParameter는 아래와 같이 타입을 지정할 수도 있다.

java -jar batch_example.jar fileName=target_data.csv baseDate(date)=2024/03/01

 

이렇게 전달된 파라미터는 JobRepository의 BATCH_JOB_EXECUTION_PARAMS 테이블에서 조회할 수 있다.

BATCH_JOB_EXECUTION_PARAMS 테이블

 

위의 설명에서 JobInstance는 Job 이름과 Job 파라미터를 기준으로 식별된다고 했다. 만약 동일한 파라미터로 동일한 Job을 실행하면 어떻게 될까?

동일한 Job 파라미터를 사용하게된다면, 동일한 JobInstance를 사용한다는 뜻일 것이다.

동일 파라미터로 Job 실행한 경우

Caused by: org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException: A job instance already exists and is complete for identifying parameters={'foo':'{value=bar, type=class java.lang.String, identifying=true}'}.
  
If you want to run this job again, change the parameters.

따라서 위와 같이 동일한 Job 이름과 파라미터로 두 번 이상 실행하는 경우 이미 JobInstance가 존재한다는 에러가 발생하게 된다.

 

그러나 식별자 역할을 하지 않는 JobParameter를 지정할 수 있다.

java -jar batch_example.jar fileName=target_data.csv -baseDate(date)=2024/03/01

위 명령어의 baseDate와 같이 접두사 '-' 를 사용한 파라미터는 JobInstance의 식별자로 사용되지 않는다.

 

JobParameters 접근하기

Job 내부에서 JobParameter에 접근하여 사용하는 방법은 여러가지가 있다.

ChunkContext를 사용하는 방법

Tasklet이 동작하는 과정에서 기본적으로 두 개의 파라미터를 가진다. 이 때 ChunkContext 인스턴스를 전달받게 되는데, 이 인스턴스는 실행 시점의 잡 상태를 표현한다.

@Bean
public Tasklet demoTasklet() {
	return (contribution, chunkContext) -> {
		String fileName = (String) chunkContext.getStepContext()
			.getJobParameters()
			.get("fileName");
        
		log.info("Demo FileName: {}", fileName);
		return RepeatStatus.FINISHED;
	};
}

위의 코드와 같이 ChunkContext를 활용하여 잡 파라미터를 획득할 수 있다. 이 경우 획득한 잡 파라미터 값은 Object 타입이므로 타입 캐스팅이 필요하다.

Late Binding을 사용하는 방법

@Bean
public Step demoStep() {
	return stepBuilderFactory.get("demoStep")
		.tasklet(demoTasklet(null))
		.build();
}

@StepScope
@Bean
public Tasklet demoTasklet(@Value("{jobParameters['fileName']}") String fileName) {
	return (contribution, chunkContext) -> {
		log.info("Demo FileName: {}", fileName);
		return RepeatStatus.FINISHED;
	};
}

위의 코드에서는 잡 파라미터를 스프링 EL을 사용하여 값을 전달한다. 이 때 잡 파라미터 값은 지연 바인딩되는데, Tasklet 빈을 생성할 때 전달받을 파라미터를 주입한다.

위 코드의 경우에는 Step의 실행 범위에 진입하는 시점에 Tasklet 빈을 생성한다. (@StepScope 사용)

(만약 지연 바인딩이 되지 않는다면, Tasklet 빈 생성은 Spring Application이 Bootstrap되는 시점이고 JobParameters를 전달받는 시점은 Job 실행을 전달받는 시점이기 때문에 위와 같이 코드를 구성할 수 없다.)

 

https://docs.spring.io/spring-batch/reference/step/late-binding.html

Late Binding of Job and Step Attributes :: Spring Batch

위 링크에 따르면 -D 매개변수를 주고 지연 바인딩을 사용할 수도 있다.

java -jar batch_example.jar fileName=target_data.csv -Dinput.file.name="file://outputs/file.txt"

@Bean
public FlatFileItemReader flatFileItemReader(@Value("${input.file.name}") String name) {
	return new FlatFileItemReaderBuilder<Foo>()
			.name("flatFileItemReader")
			.resource(new FileSystemResource(name))
			...
}

그러나 JobRegistryBackgroundJobRunner를 사용하는 경우에는 Spring Application 기동 시점에 시스템 매개변수로 전달해야 하므로 꼭 필요한 경우에만 사용하면 될 것 같다.

 

JobParameters 유효성 검증하기

JobParameters의 유효성을 검증하기 위해 Spring Batch에서 제공하는 간단한 방법은 Job 내에 JobParametersValidator를 구현하고 구성하는 것이다.

public class CustomParameterValidator implements JobParametersValidator {

	@Override
	public void validate(JobParameters parameters) throws JobParametersInvalidException {
		String fileName = parameters.getString("fileName");

		if (!StringUtils.hasText(fileName)) {
			throw new JobParametersInvalidException("fileName 파라미터가 존재하지 않습니다.");
		}
		else if (!StringUtils.endsWithIgnoreCase(fileName, ".csv")) {
			throw new JobParametersInvalidException("fileName 파라미터는 csv 확장자여야 합니다.");
		}
	}
}

validate(...) 메서드는 반환 타입이 void이기 때문에 예외가 발생하지 않으면 유효성이 검증된 것으로 볼 수 있다.

 

Spring Batch는 DefaultJobParametersValidator를 제공하고 있어, 파라미터 존재 여부에 대한 검증만을 손쉽게 수행할 수도 있다. 그러나 더 세세한 검증은 위 코드와 같이 JobParametersValidator를 구현하는 방법을 사용한다.

@Bean
public CompositeJobParametersValidator validator() {
	DefaultJobParametersValidator defaultJobParametersValidator = 
		new DefaultJobParametersValidator(new String[] {"fileName"}, new String[] {"name"});
	defaultJobParametersValidator.afterPropertiesSet();

	CompositeJobParametersValidator validator = new CompositeJobParametersValidator();
	validator.setValidators(
		Arrays.asList(new CustomParameterValidator(), defaultJobParametersValidator)
	);
    
    return validator;
}

@Bean
public Job job() {
	return this.jobBuilderFactory.get("job")
		.start(step())
		.validator(validator())
		.build();
}

위의 코드에서는 Spring Batch에서 제공하는 CompositeJobParametersValidator를 사용하여, 여러 개의 JobParametersValidator를 중첩하여 적용하였다.

 

JobParameters 증가 시키기

동일한 식별 잡 파라미터로 여러 번 잡을 실행하는 경우 예외가 발생한다. 이러한 현상을 해소하기 위해 JobParametersIncrementer를 사용할 수 있다.

JobParametersIncrementer는 Job에서 사용할 파라미터를 고유하게 생성할 수 있도록 Spring Batch에서 제공하는 인터페이스이다. RunIdIncrementer는 기본적으로 제공되는 구현 클래스이며, long 타입의 run.id 파라미터를 자동으로 증가시킨다.

@Bean
public Job job() {
	return this.jobBuilderFactory.get("job")
		.start(step())
		.validator(validator())
		.incrementer(new RunIdIncrementer())
		.build();
}

run.id 파라미터로 실행된 이력

따라서 JobRepository에 적재되는 데이터에도 run.id 값으로 실행된 이력을 확인할 수 있다.

 

JobListener 활용하기

Job 생명주기의 특정 시점에 로직을 추가할 수 있는 방법으로 Spring Batch에서는 JobExecutionListener를 제공한다.

JobExecutionListener는 beforeJob/afterJob 메서드를 제공하여 Job 이전/이후에 수행할 로직을 정의할 수 있다.

public class JobLoggerListener implements JobExecutionListener {

	@Override
	public void beforeJob(JobExecution jobExecution) {
		log.info("{} Job이 실행됨", jobExecution.getJobInstance().getJobName());
	}

	@Override
	public void afterJob(JobExecution jobExecution) {
		log.info(
			"{} Job이 {} 상태로 종료됨", 
			jobExecution.getJobInstance().getJobName(),
			jobExecution.getStatus());
	}
}


@Bean
public Job job() {
	return this.jobBuilderFactory.get("job")
		.start(step())
		.validator(validator())
		.incrementer(new RunIdIncrementer())
		.listener(new JobLoggerListener())
		.build();
}

 

위와 같이 인터페이스를 직접 구현하는 방식이 아닌 어노테이션을 활용한 방식을 사용할 수 있다.

public class JobLoggerListener {

	@BeforeJob
	public void beforeJob(JobExecution jobExecution) {
		log.info("{} Job이 실행됨", jobExecution.getJobInstance().getJobName());
	}

	@AfterJob
	public void afterJob(JobExecution jobExecution) {
		log.info(
			"{} Job이 {} 상태로 종료됨", 
			jobExecution.getJobInstance().getJobName(),
			jobExecution.getStatus());
	}
}


@Bean
public Job job() {
	return this.jobBuilderFactory.get("job")
		.start(step())
		.validator(validator())
		.incrementer(new RunIdIncrementer())
		.listener(JobListenerFactoryBean.getListener(new JobLoggerListener()))
		.build();
}

@BeforeJob, @AfterJob 어노테이션을 활용해서 더욱 간단하게 구현할 수 있게 되었고, 이에 따라 Spring AOP로 JobExecutionListener의 Proxy 객체를 생성하는 것으로 보인다.

따라서 Listener를 등록할 때에도 JobLoggerListener 객체가 아닌 프록시 객체를 JobListenerFactoryBean을 통해 찾아서 등록한다.

AbstractListenerFactoryBean에서 Proxy Object를 반환하는 과정

 

ExecutionContext 알아보기

배치 처리는 특성상 상태를 가지고 있는데 이 상태를 통해 어떤 스텝이 실행되고 있으며 어떤 레코드가 얼만큼 처리되었는지 알 수 있다.

Job이 실행될 때 생성되는 JobExecution 인스턴스 내부의 ExecutionContext에 잡의 상태가 저장된다.

JobExecution은 여러 개의 StepExecution을 가지는데 각 StepExecution도 ExecutionContext를 가진다.

(따라서 하나의 JobExecution은 여러 개의 ExecutionContext를 가지는 구조가 된다)

ExecutionContext는 웹 개발 환경의 세션과 유사하게 사용할 수 있어서, 내부에 특정 정보를 저장할 수도 있고 꺼내어 사용할 수도 있다.

public RepeatStatus execute(StepContribution step, ChunkContext context) throws Exception {
	String name = (String) context.getStepContext()
		.getJobParameters()
		.get("name");

	ExecutionContext jobContext = context.getStepContext()
		.getStepExecution()
		.getExecutionContext();

	jobContext.put("user.name", name);

	return RepeatStatus.FINISHED;
}

또한 이 상태 값들은 JobRepository의 BATCH_JOB_EXECUTION_CONTEXT 테이블에 저장된다.

인코딩되어서 저장되어 있는 상태 값들

 

다음 글: https://gojs.tistory.com/19

Spring Batch의 Step 알아보기