Spring Batch의 JobRepository 알아보기

이전 글: https://gojs.tistory.com/19

Spring Batch의 Step 알아보기

 

JobRepository?

JobRepository란 Spring Batch에서 제공하는 인터페이스로, 해당 인터페이스를 구현함으로써 배치 정보를 저장할 수 있다.

JobRepsitory를 구성하는 방법은 크게 두 가지로 구분할 수 있다. 그 중 하나는 인메모리 저장소이고, 나머지 하나는 관계형 데이터베이스 저장소이다.

 

관계형 데이터베이스 사용하기

관계형 데이터베이스를 사용하여 JobRepsitory를 구성하는 경우 아래와 같은 스키마가 구성된다.

Spring Batch 스키마 구성 (출처: https://docs.spring.io/spring-batch/reference/schema-appendix.html)

 

위 테이블 구조에서 가장 먼저 시작되는 지점은 BATCH_JOB_INSTANCE 테이블이다.

특정 Job의 name과 특정 parameter를 기준으로 하나의 레코드가 적재되며 이 레코드는 Job의 논리적인 실행을 의미한다.

그럼 아래의 BATCH_JOB_INSTANCE 테이블의 컬럼별 설명을 참고하자.

컬럼	설명
JOB_INSTANCE_ID	기본 키
VERSION	낙관적 락에 사용되는 컬럼
JOB_NAME	실행된 Job의 이름
JOB_KEY	Job 이름과 Job 파라미터의 해시 값 (JobInstance를 고유하게 식별하는데 사용되는 값)

 

BATCH_JOB_EXECUTION 테이블은 실제 Job의 실행에 대한 레코드를 적재하는 테이블이다.

Job을 실행하는 시점에 새로운 레코드가 적재되고, Job이 실행되는 중에 상태 정보가 실시간으로 갱신된다.

컬럼	설명
JOB_EXECUTION_ID	기본 키
VERSION	낙관적 락에 사용되는 컬럼
JOB_INSTANCE_ID	BATCH_JOB_INSTANCE 테이블을 참조하는 외래 키
CREATE_TIME	레코드 생성 시간
START_TIME	잡 실행 시작 시간
END_TIME	잡 실행 끝 시간
STATUS	잡 실행 상태
EXIT_CODE	잡 실행 종료 코드
EXIT_MESSAGE	잡 실행 종료 메세지
LAST_UPDATED	레코드 최종 갱신 시간

 

BATCH_JOB_EXECUTION_CONTEXT 테이블은 ExecutionContext에 대한 정보를 저장하는 테이블이다.

컬럼	설명
JOB_EXECUTION_ID	기본 키
SHORT_CONTEXT	SEREALIZED_CONTEXT 값을 트림처리한 데이터
SEREALIZED_CONTEXT	직렬화된 ExecutionContext

기본적으로 Spring 기반의 프로젝트는 Jackson 라이브러를 활용하여 JSON을 핸들링한다.

이처럼 ExecutionContext 정보를 JSON 형태로 직렬화해서 BATCH_JOB_EXECUTION_CONTEXT 테이블에서 관리한다.

 

BATCH_JOB_EXECUTION_PARAMS 테이블은 잡이 실행될 때 마다 사용된 잡 파라미터를 저장한다.

이 정보들은 Job이 재실행되는 시점에 활용되기도 하며, Job의 식별 정보 파라미터들을 다시 전달한다.

컬럼	설명
JOB_EXECUTION_ID	기본 키
TYPE_CODE	파라미터 value 타입 코드
KEY_NAME	파라미터 key 이름
STRING_VAL	파라미터 value type이 String인 경우의 값
DATE_VAL	파라미터 value type이 Date인 경우의 값
LONG_VAL	파라미터 value type이 Long인 경우의 값
DOUBLE_VAL	파라미터 value type이 Double인 경우의 값
IDENTIFYING	파라미터가 식별되는지 여부를 나타내는 플래그 컬럼

 

JobRepository에서는 Step에 관련된 정보 역시 저장하여 관리한다.

그 중 StepExecution에 대한 정보를 관리하는 BATCH_STEP_EXECUTION 테이블이 있다.

컬럼	설명
STEP_EXECUTION_ID	기본 키
VERSION	낙관적 락에 사용되는 컬럼
STEP_NAME	스텝 이름
JOB_EXECUTION_ID	BATCH_JOB_EXECUTION 테이블을 참조하는 외래 키
START_TIME	스텝 실행 시작 시간
END_TIME	스텝 실행 끝 시간
STATUS	스텝 실행 상태
COMMIT_COUNT	커밋된 트랜잭션 수
READ_COUNT	읽은 아이템 수
FILTER_COUNT	ItemProcessor가 null을 반환하여 Filtering된 아이템 수
WRITE_COUNT	쓴 아이템 수
READ_SKIP_COUNT	ItemReader에서 예외 발생하여 건너뛴 아이템 수
PROCESS_SKIP_COUNT	ItemProcessor에서 예외 발생하여 건너뛴 아이템 수
WRITE_SKIP_COUNT	ItemWriter에서 예외발생하여 건너뛴 아이템 수
ROLLBACK_COUNT	롤백된 아이템 수
EXIT_CODE	스텝 실행 종료 코드
EXIT_MESSAGE	스텝 실행 종료 메세지
LAST_UPDATED	레코드 최종 갱신 시간

 

JobExecution 내에 ExecutionContext가 존재하듯이 StepExecution에도 ExecutionContext가 존재하며, 이를 관리하는 테이블 역시 존재한다.

BATCH_STEP_EXECUTION_CONTEXT 테이블에서는 스텝 수준에서의 실행 상태를 저장한다.

컬럼	설명
STEP_EXECUTION_ID	기본 키
SHORT_CONTEXT	SEREALIZED_CONTEXT 값을 트림처리한 데이터
SEREALIZED_CONTEXT	직렬화된 ExecutionContext

 

인메모리 저장소 활용하기

Spring Batch에서 외부 데이터베이스 저장소를 사용하는 것보다 인메모리 환경의 저장소를 사용하는 것이 더 유용한 경우가 있다.

따라서 Spring Batch에서는 Map 객체를 데이터 저장소로 활용하는 JobRepository 구현체를 제공한다.

(실제 운영 시에는 관리되지 않기 때문에 사용하지 않으며, H2와 같은 인메모리 데이터베이스를 사용하는 것이 더욱 바람직하다)

 

JobRepository 추가 구성하기

@EnableBatchProcessing 어노테이션을 활용하면 Spring Batch가 제공하는 JobRepository를 활용할 수 있다.

그러나 JobRepository를 커스터마이징해서 사용해야하는 경우라면 BatchConfigurer 인터페이스를 구현하는 방법이 있다.

@EnableBatchProcessing 어노테이션을 사용하면 SimpleBatchConfiguration에서 BatchConfigurer 인터페이스를 구현한다.

구현한 인스턴스가 ApplicationContext에 올라가는 시점에 필요한 DataSource, TransactionManager 등등의 필요한 컴포넌트도 자동 구성한다.

그렇다면 BatchConfigurer를 직접 구현함으로써 이와 같은 구성들을 바꿔서 사용할 수 있다!

 

BatchConfigurer 인터페이스 명세 (출처: https://docs.spring.io/spring-batch/docs/5.0.0-M4/api/org/springframework/batch/core/configuration/annotation/BatchConfigurer.html)

위와 같이 BatchConfigurer를 명세에 맞춰 구현할 수 있다.

그러나 실제 모든 구성을 바꾸는 경우는 흔하지 않다. 따라서 DefaultBatchConfigurer에서 이미 구현되어 있는 로직을 그대로 사용하고, 필요한 부분만 오버라이딩하는 것이 더 간편한 방법이다.

 

DefaultBatchConfigurer 명세 (출처: https://docs.spring.io/spring-batch/docs/5.0.0-M4/api/org/springframework/batch/core/configuration/annotation/DefaultBatchConfigurer.html)

 

DefaultBatchConfigurer를 사용하는 경우에는 "create"를 prefix로 가지는 메서드를 오버라이딩한다.

예를 들어 getJobRepository() 메서드에서는 createJobRepository() 메서드를 호출하여 JobRepository를 만들어 반환한다.

따라서 실제로 구성할 내용에 대해서는 createJobRepository를 구현하여 구성한다.

 

public class CustomBatchConfigurer extends DefaultBatchConfigurer {

	@Autowired
	@Qualifier("repositoryDataSource")
	private DataSource dataSource;
    
	@Overrid
	protected JobRepository createJobRepository() throws Exception {
		JobRepositoryFactoryBean factoryBean = new JobRepositoryFactoryBean();
        
		factoryBean.setDatabaseType(DatabaseType.MYSQL.getProductName());
		factoryBean.setTablePrefix("CUSTOM_BATCH");
		factoryBean.setIsolationLevelForCreate("ISOLATION_REPREATABLE_READ");
		factoryBean.setDataSource(dataSource);
		
		factoryBean.afterPropertiesSet();
		
		return factoryBean.getObject();
	}
}

위의 코드와 같이 DefaultBatchConfigurer를 상속해 createJobRepository를 오버라이딩해서 커스터마이징한 JobRepository를 등록하여 사용할 수 있다.

(보통 두 개 이상의 DataSource를 구성하여 사용하는 프로젝트에서 많이 사용되는 방법이다)

위 코드에서는 테이블 Prefix, 고립 레벨, DataSource 빈 오브젝트 등을 설정하였으나 아래와 같이 더 많은 내용을 설정할 수도 있다.

접근자 이름	설명
setClobType	CLOB 컬럼에 사용할 타입
setSerializer	ExecutionContext를 직렬화/역직렬할 때 사용하는 ExecutionContextSerializer
setLobHandler	Oracle 예전 버전에서 LOB을 취급하는 LobHandler 구성
setMaxVarCharLength	실행 컨텍스트, 종료 메세지 등 길이 자르는데 사용 (설정X)
setDataSource	DataSource 설정
setJdbdOperations	JdbeOperaions 설정
setDatabaseType	데이터베이스 타입 설정
setTablePrefix	"BATCH_"라는 접두어 대신 사용할 접두어
setIncrementerFactory	테이블 기본 키 증분하는데 사용되는 증분기
setValidateTransactionState	기존 트랜잭션있는지 여부에 대한 플래그
setIsolationLevelForCreate	트랜잭션 고립 레벨 설정 (Default Serializble)
setTransactionManager	여러 데이터베이스를 사용하는 경우 2단계 커밋을 지원하는 TransactionManager 지정

 

JobExplorer 사용하기

JobExplorer는 JobRepository의 저장소에서 동일한 데이터를 읽기 전용으로 보는 뷰이다.

또한, 이력 데이터나 최신 데이터에 접근하는 시작점으로, 아래의 메서드를 사용자가 사용할 수 있도록 제공한다.

메서드	설명
findRunningJobExecutions(String jobName)	종료 시간이 없는 모든 JobExecution 반환
findJobInstancesByName(String name, int start, int end)	전달받은 이름을 가진 JobInstance 목록 반환 (페이징)
getJobExecution(Long executionId)	전달받은 ID를 가진 JobExecution 반환
getJobExecutions(JobInstance instance)	전달받은 JobInstance과 관련된 모든 JobExecution 목록 반환
getJobInstance(Long instanceId)	전달받은 ID를 가진 JobInstance 반환
getJobInstances(String jobName, int start, int end)	전달받은 인덱스부터 범위 내에 있는 JobInstance 목록 반환
getJobInstanceCount(String jobName)	전달받은 잡 이름으로 생성된 JobInstance 개수 반환
getJobNames()	모든 잡 이름을 알파벳 순서로 반환
getStepExecution(Long jobExecutionId, Long stepExecutionId)	전달받은 ID들을 기준으로 StepExecution 반환

 

그렇다면 해당 메서드들을 활용해서 JobExplorer를 사용해보자.

Job, Step, Tasklet 구성하고 JobExplore 의존성 주입받아 파라미터로 전달

전달받은 JobExplore를 가지고 필요한 데이터 조회하여 처리

정상적으로 실행된 Tasklet 로직 (JobExplore 활용)

위 코드에서 RunIdIncrementer를 사용해서 Job을 두 번 실행하였기 때문에 JobInstance가 두 개 생성되었다.

그리고 실행 중인 JobExecution의 ExitStatus는 UNKNOWN, 종료된 JobExecution의 ExitStatus는 COMPLETED로 정상 출력된 것을 확인할 수 있다.