Spring Batch - 개요, 도메인 이해

최근에 이직한 회사에서 스프링 배치를 사용해서, 스프링 배치를 조금 더 잘 활용하기 위해 공부를 시작했다. 

 

시중의 책, 강의, 공식 문서를 기반으로 내용을 정리한다. 

실무에서 참고할만한 상황이 있다면 해당 사항도 같이 작성할 예정이다.

---

스프링 배치 개요

• 자바 기반 표준 배치 기술 부재로 만들어짐.
• SpringSouce 와 Accenture의 합작품
• Accenture는 이전에 소유했던 배치 처리 아키텍처 프레임워크를 Spring Batch 프로젝트에 기증함

배치 핵심 패턴

1. Read - 데이터 조회 (extract)
2. Process - 특정 방법으로 데이터를 가공 (transform)
3. Write - 데이터를 수정된 양식으로 저장 (load)

배치 시나리오

1. 배치 프로세스를 주기적으로 커밋
2. 동시 다발적인 job의 배치 처리 (다중 스레드 - 병렬 처리)
3. 실패 후 수동 또는 스케줄링에 의한 재시작
4. 의존 관계로 이루어져있는 step을 여러개 순차적으로 처리한다
5. 조건적 flow 구성 (조건에 따라 flow를 관리함.)
6. 반복하고, 재시도하고(장애가 생기면 재시도하는 것), skip(예외가 발생하면 skip처리하는 기능)하는 처리도 있음.

아키텍처

• Application: 개발자가 만든 모든 배치job과 커스텀 코드를 포함
  • 개발자는 업무로직의 구현에만 집중하고 공통적인 기반 기술은 프레임웍이 담당하게 함
• BatchCore :
  • job을 실행, 모니터링, 관리하는 api로 구성
  • JobLauncher클래스, Job, Step, Flow가 있음
• Batch Infrastructure :
  • Application, Core 모두 공통 Infrastrucuture위에서 빌드함
  • JOB의 실행의 흐름과 처리를 위한 틀을 제공함
  • Reader, Processor Writer, Skip, Retry 등이 속함.

Job 

• 배치 계층 구조에서의 가장 상위에 있는 개념으로, 하나의 배치 작업자체를 의미
• Job Configuration을 통해 생성되는 객체 단위
• 배치 Job을 구성하기 위한 최상위 인터페이스로, 스프링 배치가 기본 구현체를 제공함
• 여러 Step을 포함하고 있는 컨테이너로, 반드시 한 개 이상의 Step으로 구성해야된다.
• job의 종류로 SimpleJob과 FlowJob이 있음

SimpleJob

• 순차적으로 Step을 실행시키는 Job
• 모든 Job에서 유용하게 사용할 수 있는 표준 기능 

FlowJob

•  특정한 조건에 따라 Step을 구성시켜, 작업을 진행하는 Job
•  Flow객체를 실행시켜서 작업을 진행함. 

JobInstance

• 고유하게 식별 가능한 작업 실행을 나타냄 
• Job이 실행될 때 생성되는 Job의 논리적인 실행 단위 객체로, 고유하게 식별 가능한 작업 실행을 나타냄
•  JobInstance 생성 및 실행
  • => 처음 시작하는 job + job parameter일 경우, 새로운 job instance생성
  • => 이전과 동일한 job + job parameter 일 경우 이미 존재하는 JobInstance 리턴
•  Job과는 1:M 관계

 

정리하자면,

1. JobLauncher가 Job을 실행하는데, job과 parameters를 받아 실행한다.

2. JobRepository는 인자로 넘겨준 Job객체, JobParameters 두 개의 값을 가지고 db로부터 확인하는 과정 지남.

 현재 실행한 Job이 처음 실행하는건지, 아니면 이전에 한번이라도 실행한 Job인지 확인. 이미 한번이라도 수행해서 존재하는 값이면 기존에 있던 JobInstance 리턴 

3. 그렇지 않다면 새로운 JobInstance 생성함.

 

JobParameter

• Job을 실행할 때 함께 포함되어 사용되는 파라미터를 가진 도메인 객체
• 하나의 Job에 존재할 수 있는 여러 개의 JobInstance를 구분하기 위한 용도
• JobParameters와 JobInstance는 1:1 관계 

JobParameter 생성 및 바인딩 하는 법

• 어플리케이션 실행 시 주입
• JobParameterBuilder, DefaultJobParametersConverter 사용 
• SpEL 이용
  • @Value("#{jobParameter[requestDAte])"), @JobScope, @StepScope 선언 

 

JobExecution

• JobInstance에 대한 한 번의 시도를 의미하는 객체로서, Job실행 중에 발생한 정보들을 가지고 있는 객체
• JobInstance의 상태 종류 
  • 1. FAILED : JOB Instance의 실행이 완료되지 않은 것으로 간주함
  • 2. COMPLETED: Job Instance실행이 완료된 것으로 간주해서 재 실행이 불가능함.
    •  JobExcecution의 실행 상태 결과가 'COMPLETED' 될 때 까지 하나의 Job Instance내에서 여러번의 시도 가능.
• JobParameters, JobInstance, BatchStatus(위의 실행 상태), ExitStatus(종료 코드), failureException(Job 실행 중 발생한 예외 리스트), startTime(Job을 실행 때의 시스템 시간), createTime(JobExcecution이 처음 저장될 때의 시스템 시간), endTime(성공 여부와 상관 없이 실행이 종료되는 시간), lastUpdated(JobExcecution이 저장될 때의 시스템시간) 을 포함함. 

Step

• Batch Job을 구성하는 독립적인 하나의 단계
• 실제 배치 처리를 하는 도메인 객체 
• 모든 Job은 하나 이상의 Step으로 구성됨. 
• TaskletStep : 가장 기본이 되는 클래스로, Tasklet 타입의 구현체를 제어함
• PartitionStep : 멀티 스레드 방식으로 Step을 여러 개로 분리해서 실행함. 

구현체로 TaskletStep, PartitionStep, JobStep, FlowStep이 있다.

•  TaskletStep: 가장 기본이 되는 클래스로, Tasklet 타입의 구현체들을 제어함

public Step taskletStep() {
	return this.stepBuilderFactory.get("step")
    							  .tasklet(myTasklet())
                                  .build();
}

 

public Step taskletStep() {
	return this.setpBuilderFactory.get("step")
    							  .<Member, Member>chunk(10)
                                  .reader(reader())
                                  .writer(writer())
                                  .build();
}

 

•  PartitionStep : 멀티 스레드 방식으로, Step을 여러 개로 분리하여 실행한다. 
• JobStep : Step내에서 Job을 실행하도록 함. 

public Step jobStep() {
	return this.stepBuilderFactory.get("step")
    							  .job(job())
                                  .launcher(jobLauncher)
                                  .parametersExtractor(jobParametersExtractor())
                                  .build();
}

 

• FlowStep: Step내에서 Flow를 실행하도록 함.

  public Step flowStep() {
  	return this.stepBuilderFactory.get("step")
    							  .flow(myFlow())
                                  .build();
  }

 

Step <- AbstractStep <- ( JobStep / TaskletStep / FlowStep / PartitionStep) 

이러한 구조 

 

StepExcecution 

• Step에 대한 한 번의 시도를 의미하는 객체로, Step실행 중에 발생한 정보를 저장하고 있는 객체
  • => 시작 시간, 종료시간, 상태, commit count, rollback count 등의 속성을 지님. 
• Step이 매번 시도될때마다 생성되며, 각 Step별로 생성된다.
• Job이 재시작되도, 이미 성공적으로 완료된 Step은 재실행되지 않고, 실패한 Step만 실행된다. 
• Step의 StepExecution이 모두 정상적으로 완료되어야, JobExcecution이 정상적으로 완료된다.
• Step의 StepExcecution중 하나라도 실패하면 JobExcecution은 실패함. 
• Job Excecution과 StepExcecution은 1:M 관계
• 하나의 Job에 여러 개의 Step으로 구성했을 경우, 각 StepExcecution은 하나의 JobExcecution을 부모로 지닌다. 

 

StepContribution

• 청크 프로세스의 변경사항을 버퍼링한 후 StepExcecution 상태를 업데이트하는 도메인 객체
• 청크 커밋 직전에 StepExcecution의 apply(contribution)메서드를 호출하여, StepExcecution에 상태를 업데이트함. 

  예컨대, TaskletStep -> StepExcecution -> StepContribution 순서로 진행하는데,  
Tasklet하위의 ItemReader, ItemProcessor, ItemWriter에서 실행된 결과 변수인 readCount, filterCount, writeCount 혹은 skipcount개수를 StepContribution이 받아서, StepExcecution이 완료되는 시점에 apply메서드를 호출해서, 속성들의 상태를 최종 업데이트함.  

 

• 기본 종료코드 외 사용자 정의 종료 코드를 만들어서, 적용할 수 있음. 

 

ExcecutionContext

• 프레임워크에서 유지 및 관리하는 키 / 값으로 된 컬렉션으로, StepExcecution 혹은 JobExcecution 객체의 상태를 저장하는 공유 객체이다.
• key, value형태로 db에 직렬화된 상태로 저장됨. 
• 각 Job의 JobExecution에 저장되며, job간 서로 공유가 되지 않고 해당 Job의 Step간 서로 공유됨 
• 이미 처리한 Row데이터를 건너 뛰고, 이후에 수행해야할 때 활용함. 

@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Excpetion {
	
    ExcecutionContext jobExcecutionContext = contribution.getStepExcecution().getJobExcecution().getExcecutionContext();
    ExcecutionContext stepExcecutionContext = contribution.getStepExcecution().getExcecutionContext();
    
    String jobName = chunkContext.getStepContext().getStepExcecution().getJobExcecution().getJobInstance().getJobName();
    String stepName = chunkContext.getStepContext().getStepExcecution().getStepName();
    
    if(jobExcecutionContext.get("jobName") == null) {
        jobExcecutioncontext.put("jobName", jobName);
    }
    
    if(stepExcecutionContext.get("stepName") == null) {
    	stepExcecutionContext.put("stepName", stepName);
    }
    
    return RepeatStatus.FINISHED;
}

 

JobRepository

•  배치 작업 중의 정보를 저장하는 저장소 역할
• job의 수행 여부, 수행 종료 시간, 결과 등 배치 작업의 수행과 관련된 모든 metaData저장함
  • => JobLauncher, Job, Step 구현체 내부에서 CRUD 기능을 처리함. 
• EnableBatchProcessing 어노테이션만 선언시, JobRepository가 자동으로 빈이 생성이 됨
• BatchConfigurer 인터페이스를 구현하거나, BasicBatchConfigurer를 상속해서 JobRepository 설정을 커스터마이징할 수 있음
  • => JDBC 방식(내부적으로 AOP기술을 통해 트랜잭션 처리를 해주고 있음) / In Memory 방식으로 설정 (MapJobRepostioryFactoryBean) => 성능등의 이유로 도메인 오브젝트를 굳이 디비에 저장하고 싶지 않을 경우나, TEST 나 프로토타입의 빠른 개발이 필요할때 사용한다.

 

JobLauncher

• 배치 Job을 실행시키는 역할을 함
•  Job과 JobParamters를 인자로 받으며, 요청된 배치 작업을 수행한 후, 최종 client에게 JobExcecution을 반환함.
• 스프링 부트 배치가 구동 되면 JobLauncher 빈이  자동 생성됨
• Job실행
  • => JobLauncher.run(Job, JobParameters) 
  • => 스프링 부트 배치에서는 JobLauncherApplicationRunner가 자동적으로 JoubLauncher를 실행시킴
• - 동기적 실행
  • => taskExcecutor를  SyncTaskExcecutor로 설정한 경우 (기본 값은 SyncTaskExcecution)
  • => 배치 처리를 최종 완료한 이후 Client에게 JobExcecution을 반환
  • ExitStatus.FINISHED or FAILED 최종 완료 후 응답 값 반환
• - 비동기적 실행 
  • => taskExcecutor가 SimpleAsyncTaskExcecutor로 설정할 경우
  • => Client에게 바로 JobExcecution을 반화하고 바로 배치 처리 완료 
  • ExitStatus.UNKNOWN 즉시 응답값 반환