Spring Batch 5 와 ???

spring batch 5는 이전 버전과 많이 변경되었다. 기본적인 스프링배치부터 스프링 배치 5.0은 어떻게 달라졌는지 직접 실무에 적용해보면서 느꼈던 경험을 적어본다.

spring batch Architecture

Architecture

공식문서에 나온 아키텍처이다. Application, Core, and Infrastructure 로 구성되어있다.

• Application: 애플리케이션에는 Spring Batch를 사용하여 개발자가 작성한 모든 배치 작업과 사용자 정의 코드가 포함되어 있다.
• Batch Core: 배치 작업을 시작하고 제어하는 데 필요한 핵심 런타임 클래스가 포함되어 있습니다. JobLauncher, Job, and Step이 포함되어있다.
• Batch Infrastructure: 애플리케이션과 코어는 모두 공통 인프라 위에 구축된다. 이 인프라에는 readers(ItemReader) 와 writers(ItemWriter) 그리고 services (such as the RetryTemplate) 와 핵심 프레임워크 자체가 포함되어 있다 .

Principle and Guidelines

• 배치 어플리케이션 내에서 가능한한 복잡한 로직은 피하고 단순하게 설계합니다.
• 데이터 처리하는 곳과 데이터의 저장소는 물리적으로 가능한한 가까운 곳에 위치하게 합니다.
• 불필요한 테이블 또는 인덱스 스캔이 발생하지 않도록 한다.
• WHERE SQL 문의 절 에 키 값을 지정해야 한다.
• 처리 시간이 많이 걸리는 작업은, 시작하기 전 충분한 메모리를 할당해서 메모리 재할당에 시간을 소모되지 않도록 한다.
• 가능한 경우 내부 검증을 위한 체크섬을 구현합니다. 예를 들어 플랫 파일에는 파일의 총 레코드와 키 필드의 집계를 알려주는 트레일러 레코드가 있어야 합니다.
• 일괄 실행에서 작업을 두 번 수행하지 마십시오. 예를 들어 보고 목적으로 데이터 요약이 필요한 경우 데이터가 처음 처리될 때 가능한 경우 저장된 총계를 늘려 보고 응용 프로그램이 동일한 데이터를 다시 처리할 필요가 없도록 해야 합니다.

더 자세한 이야기는 공식 문서에 적혀있다.

메타 테이블

마찬가지로 여러 블로그에 남아있어서 간단하게 기록한다.

BATCH_JOB_EXECUTION

• END_TIME: 실행이 종료된 시점을 TIME_STAMP로 기록하며, 실행 도중 오류 발생시 기록이 안될 수 있다.
• EXIT_MESSAGE: 실패했을때의 메시지로 어디서 어떤 예외가 터졌는지 trace가 남는다.

BATCH_JOB_EXECUTION_CONTEXT

• SHORT_CONTEXT: 공유할 데이터를 넣어놓는 Map이다. base64로 인코딩되어 있다.

BATCH_STEP_EXECUTION

• END_TIME: 실행이 종료된 시점을 TIME_STAMP로 기록하며, 실행 도중 오류 발생시 기록이 안될 수 있다.
• read count, write count 및 commit count를 기록한다. 이를 이용해 retry, 예외 처리에서 사용한다.

JOB_INSTANCE는 하나만 생성되고 JOB_EXECUTION은 JOB이 실행될때마다 생성된다.(1:N 관계)

이외에도 여러 테이블이 있다.(총 9개)

Chunk 아키텍처

chunk architecture

• ItemProcessor가 iterator를 돌면서 처리한다. 따라서 processor에서 I/O작업은 없애는 것을 추천한다.
• chunkSize와 paging사이즈가 같아야, chunk사이즈 검사를 한번에 통과할 수 있다. 다르면 반복되기 때문에 같아야 한다.
• IteamReader, ItemProcessor는 Chunk내 개별 아이템을 처리한다. ItemWriter는 Chunk크기만큼 일괄 처리한다.

Cursor vs Paging ItemReader

• cursor: 커넥션이 한번 열리면, 배치가 끝날때까지 열리므로 db connection time이 중요하다.

Cursor 방식으로 데이터를 읽어서 처리하는 경우 읽을 데이터가 존재하지 않을 때까지 DB 커넥션을 유지한채로 계속 데이터를 FetchSize 만큼 가지고 와서 한 건씩 read 하게 됩니다.
그렇기 때문에 스트리밍 방식으로 계속 데이터를 메모리에 가지고 와서 처리하기 때문에 한번의 쿼리 결과로 조회된 데이터를 모두 처리할 때까지 커넥션이 유지되고 메모리에 적재되고 건건히 처리되는 방식입니다.
물론 GC 가 적절한 시점에 메모리 정리를 하겠지만 커넥션이 이루어지고 닫히기 까지 하나의 트랜잭션 안에서 모든 데이터가 처리되기 때문에 메모리에 계속 할당한다고 볼 수 있다고 한다.
참고로 Cursor 방식은 내부적으로 스냅샷 방식으로 동작하기 때문에 메모리 사용량이 많아지기도 한다고 한다.
대신 페이징 방식은 커넥션이 이루어지고 쿼리를 실행한 후 페이징 크기 만큼 데이터를 처리하고 커넥션이 닫힙니다. 그리고 다시 커넥션이 이루어지고 쿼리를 새로 실행하는 방식으로 이루지기 때문에 페이징 크기만큼 생성된 트랜잭션 안에서 메모리 처리가 이루어진다고 볼 수 있습니다.
출처: https://www.inflearn.com/questions/450829/cursor-%EB%B0%A9%EC%8B%9D-itemreader-%EA%B4%80%EB%A0%A8-%EC%A7%88%EB%AC%B8%EC%9E%85%EB%8B%88%EB%8B%A4

• paging: 한 페이지 읽을때마다 커넥션을 끄고 키므로 대용량에 더 효율적이다.

JpaCursorItemReader는 올바른 MySQL Cursor 방식이 아닙니다. 데이터를 DB에서 모두 읽고 서비스 인스턴스에서 직접 Iterator로 cursor로 동작하는 것처럼 흉내 내는 방식입니다. 즉, 모든 데이터를 메모리에 들고 있기 때문에 OOM을 유발합니다.

 

사이즈가 작을때 paging을 사용하면, 매번 커넥션을 맺고 연결하는 작업이 더 비쌀테니 이럴때는 cursor를 사용하는게 좋을 것 같다.(keep-alive가 나온 이유에서부터 생각해봤다)

ItemStream

• open
  • ExecutionContext에서 제공된 strem을 open한다. 저장된 실행상태를 읽어온다.
• update
  • 변화된 상태를 ExecutionContext에 저장한다.
• close
  • stream을 닫는다.

ItemReader나 ItemWriter를 커스텀하게 활용하려면 해당 인터페이스를 구현해야. 재시도시에 실패한 시점부터 잡을 실행시킬 수 있다.

JobLauncherApplicationRunner

CommandLineJobRunner vs jobLauncher

spring boot web이 따로 없을때는 CommandLineJobRunner는 java -jar를 실행했을때 이용할 수 있다.

반면에 web이 있으면 jobLauncher를 사용하여 job을 실행시킬 수 있다. 또한 비동기로 돌아간다고 한다.

Asynchronous Job Launcher Sequence From Web Container

job이 실행중인지 확인하려면 jobExplorer에서 job이 실행중인지 확인할 수 있다.

        Set<JobExecution> runningJobExecutions = jobExplorer.findRunningJobExecutions(jobName);

Job Incrementer()

web환경에서 jobLauncher를 통해서 실행 jobLauncher.run(job, jobParameters);하면, job실행 이전에 jobParam의 설정을 마친다. 따라서 추가로 custom한 increment api가 동작하지 않는다. 따라서 addString("run.id", new Date()) 처럼 직접 넣어주면 된다.

DB설정

• rewriteBatchedStatments=true
  • Connector/J가 쿼리를 서버에 개별적으로 제출하지 않으려고 하기 때문. 따라서 쿼리 중 하나가 실패하면 후속 쿼리가 실행되지 않는다. Mysql서버가 한번에 모아서 쿼리를 날리도록 해준다.
  • 따라서 쿼리 중 하나가 실패하면 후속 쿼리가 실행되지 않습니다. 반면 rewriteBatchedStatements=false를 사용하면 실패에 관계없이 모든 쿼리가 실행됩니다.

Spring Batch 5.x

대부분의 블로그들은 스프링 배치 5.0 이전 버전을 기준으로 설명되어있다. 따라서 추가적인 부분만 적어본다.

1. BatchAutoConfiguration (DefaultBatchConfiguration & EnableBatchProcessing)

이전에는 @EnableBatchProcessing 어노테이션을 통해서 스프링 배치의 스프링 부트 자동설정을 활성화할 수 있었다. 하지만 이제는 스프링 부트의 자동설정을 사용하기 위해서는 삭제해야한다.

@EnableBatchProcessing 명시하는 방법 또는 DefaultBatchConfiguration 을 상속하여 활성화되는 빈은 이제 스프링 부트의 자동설정을 밀어내고(back-off), 애플리케이션의 설정을 커스텀하는 용도로 사용된다.

따라서 @EnableBatchProcessing 이나 DefaultBatchConfigration 을 사용하면 spring.batch.jdbc.initialize-schema 등의 기본 설정이 동작하지 않는다. 또한 부트를 실행시킬 때 Job 이 자동으로 실행되지 않으므로 Runner 의 구현이 필요하다.

• 스프링 배치가 초기화 될 때 자동으로 실행되는 설정 클래스
• Job 을 수행하는 JobLauncherApplicationRunner 빈을 생성
• 5.0부터는 @EnableBatchProcessing, DefaultBatchConfituration을 설정하면 @ConditionalOnMissingBean에 의해 JobLauncherApplicationRunner가 실행되지 않는다는 점을 꼭 기억하자

* @ConditionalOnMissingBean 은 동명의 빈이 정의되어있으면 해당 동명의 빈을 사용하고 아니면 현재 빈을 사용한다. 따라서 DefaultBatchConfiguration을 상속받으면 jobLauncherApplicationRunner도 구현해줘야 한다.

@AutoConfiguration(after = { HibernateJpaAutoConfiguration.class, TransactionAutoConfiguration.class })
@ConditionalOnClass({ JobLauncher.class, DataSource.class, DatabasePopulator.class })
@ConditionalOnBean({ DataSource.class, PlatformTransactionManager.class })
@ConditionalOnMissingBean(value = DefaultBatchConfiguration.class, annotation = EnableBatchProcessing.class)
@EnableConfigurationProperties(BatchProperties.class)
@Import(DatabaseInitializationDependencyConfigurer.class)
public class BatchAutoConfiguration {

	@Bean
	@ConditionalOnMissingBean
	@ConditionalOnProperty(prefix = "spring.batch.job", name = "enabled", havingValue = "true", matchIfMissing = true)
	public JobLauncherApplicationRunner jobLauncherApplicationRunner(JobLauncher jobLauncher, JobExplorer jobExplorer,
			JobRepository jobRepository, BatchProperties properties) {
		JobLauncherApplicationRunner runner = new JobLauncherApplicationRunner(jobLauncher, jobExplorer, jobRepository);
		String jobNames = properties.getJob().getName();
		if (StringUtils.hasText(jobNames)) {
			runner.setJobName(jobNames);
		}
		return runner;
	}
    ...
 }

2. mutlple job(다중 잡)

1. 이제 단일 Job 을 감지하면 부트가 실행될 때 Job 을 실행시킵니다. 만약 여러 개의 Job 이 context 에 존재한다면, 부트를 실행할 때 spring.batch.job.name 을 통해 실행시킬 Job 을 명시해줘야 한다.
   1. 그렇지 않으면 Caused by: java.lang.IllegalArgumentException: Job name must be specified in case of multiple jobs 이러한 에러를 마주한다.
2. spring-configuration-metadata.json 을 보면 job.enabled는 default가 true인데, 모든 jobs들을 실행시킨다. 그러므로 job.name을 반드시 명시해주자.
3. 참고로 job.name={job1}, {job2} 이렇게 이름을 주면 job을 찾지 못한다. 하나의 job을 명시해주자

enabled

3. JobParameter 지원 범위 확대

library

v4 에서의 스프링 배치는 Job parameter 로 Long, String, Date, Double 만 사용이 가능했었다.

v5 에서는 여기에 더해 converter 를 직접 구현하는 것으로 모든 종류의 타입을 JobParameter 로 사용할 수 있도록 개선되었습니다. 또한 내장으로 LocalDateTime부터 LocalDate, Time까지 Converter가 들어가있으니 그냥 사용하기만 하면 된다.

Q1. chunk size는 왜 job Parameter로 받지 않을까?

사실 안받는게 아니라 못받는다. (제가 테스트 해봤을땐 그랬습니다..)

@Bean
@JobScope
public Step step1() {
        return new StepBuilder("step1", jobRepository)
        .<PointsStatistics, PointsStatistics>chunk(jobParameter.getChunkSize(), platformTransactionManager)
        .reader(reader())
        .writer(writer())
        .build();
}

위처럼 job param을 통해서 chunk size를 런타임에 주입하면 null이 들어왔다고 에러가 나온다. 실제 디버깅을 해보면 아래의 BeanCreateionException 에러 메시지가 나온다.

BCE

StepBuilder의 구성을 보면 chunkSize를 입력해야하는데 이때, null이 들어간 상태로 만들어지기때문에 정상적으로 step빈이 만들어지지 않는 것이다.

step Builder

그렇다면 Step빈이 생성되는 시점이 jobParam의 late binding시점에 생성되도록 하면 더 편하지 않을까? 라는 의문을 갖게 되었다.

아래의 글을 보고 이해를 했다. 해석하자면 아래와 같다.

I mentioned that the chunk size could be made dynamic through application/system properties or job parameters. I deliberately put application/system properties first because I would recommend this way to configure such "technical" properties (chunk-size, thread-pool size, database connection pool size, etc). I would use job parameters for "business" properties like input file name, input table name, etc.

job param이외에 application/system properties도 받을 수 있도록 의도적으로 두었다고한다. 이 이유가

"technical" properties( chunk-size, thread-pool size, database connection pool size, etc)은 job param으로 받지 않고, "business" properties like input file name, input table name, etc. 와 같은 것들을 job param으로 받도록 하기 위함이라고 한다.

따라서 chunk사이즈를 설정한 후에 job Param이 생성되고 이후부터 jobParam이 주입되어 사용될 수 있다. 그러므로 chunkSize를 설정할 때, jobParam을 주입할 수 없다. (하지만 편볍으로 설정할수 있다곤 한다.)

출처: Dynamic Commit interval support through StepExecution class. · Issue #4134 · spring-projects/spring-batch

 

Reference

https://www.inflearn.com/course/%EC%8A%A4%ED%94%84%EB%A7%81-%EB%B0%B0%EC%B9%98/dashboard

https://techblog.woowahan.com/2695/

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