Spring Batch에서 Job과 Step🛠

문지은·2022년 1월 6일

spring batch 스프링 배치 완벽 가이드

Job

독립적으로 실행할 수 있는 고유하며 순서가 지정된 스텝의 목록

Job runner

잡의 실행은 Job runer에서 시작된다.

CommandLineJobRunner : 스크립트 or 명령행 이용
JobRegistryBackgroundJobRunner : 자바 프로세스 내에서 쿼츠나 JMX 후크 같은 스케줄러 이용
JobLauncherCommandLineRunner : Spring Boot 서버가 올라갈 때 모든 Job 타입의 빈을 실행

Job -> JobInstance -> JobExecution
JobInstanace는 성공적으로 수행된 JobExecution이 있다면 완료된 것으로 간주. 한번 완료되면 다시 실행 불가

Job 구성하기

application.yml 파일

spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.url=jdbc:mysql://localhost:3306/batch?serverTimezone=UTC&characterEncoding=UTF-8
spring.datasource.username=root
spring.datasource.password=

# spring batch와 관련된 table들이 rdbms에 자동으로 생성
spring.batch.jdbc.initialize-schema=always

java 파일

// spring batch에 필요한 bean들 autowired 가능
@EnableBatchProcessing
@SpringBootApplication
public class SpringBatchApplication
{

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public Step step1()
    {
    	// step이 FINISHED 상태로 완료되어야 성공으로 인식
        return this.stepBuilderFactory.get("step1")
            .tasklet(
            	// stepContribution : 아직 커밋되지 않은 현재 트랜잭션에 대한 정보
                // chunckContext : 실행 시점의 job 상태 제공
                ((stepContribution, chunkContext) -> {
                    System.out.println("Hello World!");
                    return RepeatStatus.FINISHED;
                })
            ).build();
    }

    // 어플리케이션 가동 시 job으로 인식되는 bean들 자동 실행
    @Bean
    public Job job()
    {
    	// step1을 포함하는 job 생성
        return this.jobBuilderFactory.get("job")
            .start(step1())
            .build();
    }

    public static void main(String[] args)
    {
        SpringApplication.run(SpringBatchApplication.class, args);
    }

}

job 실행 -> step1 실행 -> hello world 출력 -> job 완료(status : COMPLETED)

COMPLETED로 job이 완료되었으므로 똑같은 job과 parameter로 실행할 경우 에러가 발생한다.

RDBMS에는 batch와 관련된 테이블들이 생성
테이블의 내용에는 실행했던 job과 step에 대한 내용들이 저장된다.

Job Parameter

Job에 parameter를 전달하는 것은 두가지 방법이 있다.

Chunck Context

    @Bean
    public Tasklet helloWorldTasklet()
    {
        return ((stepContribution, chunkContext) -> {

            // chunckContext를 통해 실행 시점의 job 상태에서 parameter 가져오기
            // map<String, Object>이므로 type casting 필요
            String name = (String) chunkContext.getStepContext()
                .getJobParameters()
                .get("name");

            System.out.printf("Hello World! %s%n", name);
            return RepeatStatus.FINISHED;
        });
    }

Late Binding

    @StepScope
    @Bean
    public Tasklet helloWorldTasklet(@Value("#{jobParameters['name']}") String name)
    {
        return ((stepContribution, chunkContext) -> {
            System.out.printf("Hello World! %s%n", name);
            return RepeatStatus.FINISHED;
        });
    }

Late Binding 방식에는 Step이나 Job Scope를 가져야 한다.

StepScope, JobScope란?
bean 생성 시점을 Step, Job 시점으로 늦춘다. 만약 서버를 올릴 때 tasklet을 한번에 모두 생성시키면 한 tasklet에 대해서 동시에 여러 step들이 실행되면서 침범당할 수 있다. 이를 방지하기 위해서 사용한다.

Job Parameter의 유효성 검사

    @Bean
    public CompositeJobParametersValidator validator()
    {
        CompositeJobParametersValidator validator = new CompositeJobParametersValidator();

        DefaultJobParametersValidator defaultJobParametersValidator = new DefaultJobParametersValidator(
            new String[]{"fileName"},
            new String[]{"name"}
        );

        defaultJobParametersValidator.afterPropertiesSet();

        // 여러 유효성 검사를 하고 싶다면 여러개의 validator를 CompositeJobParametersValidator에 추가
        validator.setValidators(
            Arrays.asList(
            	// 미리 만들어둔 ParamerterValidator
                new ParameterValidator(),
                // 위에서 만든 defaultJobParametersValidator
                defaultJobParametersValidator
            )
        );

        return validator;
    }

parameter 이용해서 job 여러번 실행하기

incrementer 사용하는 방법

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

job repository에서 BATCH_JOB_EXECUTION_PARAMS을 살펴보면 run.id가 증가하고 있는 걸 확인할 수 있다!

그 외에도 DailyJobTimestamper를 이용해서 현재 시간을 이용해서 잡을 반복 실행할 수도 있다.

현재 우리팀 코드를 살펴보니 incrementer를 사용하진 않고 현재 시간으로 잡 파라미터를 생성해서 넘겨주고 있었다.

public void runJob(Job targetJob) throws JobParametersInvalidException,
        JobExecutionAlreadyRunningException,
        JobRestartException, JobInstanceAlreadyCompleteException
    {
        try
        {
        // 현재 시간을 parameter에 추가함으로써 여러번 실행이 가능하도록 함
            JobParameters params = new JobParametersBuilder()
                .addString("JobID", String.valueOf(System.currentTimeMillis()))
                .toJobParameters();
            jobLauncher.run(targetJob, params);
        }
        catch (Exception e)
        {
            log.info("{} job 스케쥴 등록에 실패했습니다.", targetJob.getName());
            throw e;
        }
    }

Job Listener

Job Listener를 이용해서 스프링 배치의 생명 주기의 여러 로직(알림, 초기화, 정리)을 추가할 수 있다. 여기에도 두가지 방법이 있다.

JobExecutionListener 인터페이스 이용

public class JobLoggerListener implements JobExecutionListener
{
    private static String START_MESSAGE = "%s is beginning execution";
    private static String END_MESSAGE = "%s has completed with the status %s";

    @Override
    public void beforeJob(JobExecution jobExecution)
    {
        System.out.printf((START_MESSAGE) + "%n", jobExecution.getJobInstance().getJobName());
    }

    @Override
    public void afterJob(JobExecution jobExecution)
    {
        System.out.printf(
            (END_MESSAGE) + "%n",
            jobExecution.getJobInstance().getJobName(),
            jobExecution.getStatus()
        );
    }
}

@BeforeJob, @AfterJob 어노테이션 이용

public class JobLoggerListener
{
    private static String START_MESSAGE = "%s is beginning execution";
    private static String END_MESSAGE = "%s has completed with the status %s";

    @BeforeJob
    public void beforeJob(JobExecution jobExecution)
    {
        System.out.printf((START_MESSAGE) + "%n", jobExecution.getJobInstance().getJobName());
    }

    @AfterJob
    public void afterJob(JobExecution jobExecution)
    {
        System.out.printf(
            (END_MESSAGE) + "%n",
            jobExecution.getJobInstance().getJobName(),
            jobExecution.getStatus()
        );
    }
}

ExecutionContext

ExecutionContext는 스프링 배치에서 job과 step에 대한 상태를 저장하고 있다. 또한 하나의 job과 step에 대해서 각각의 ExecutionContext를 가진다.

ExecutionContext 조작하기

public class HelloWorld implements Tasklet
{
    public static final String HELLO_WORLD = "Hello World, %s";

    @Override
    public RepeatStatus execute(StepContribution stepContribution, ChunkContext chunkContext) throws Exception
    {
        String name = (String) chunkContext.getStepContext()
            .getJobParameters()
            .get("name");

	// ExecutionContext 접근
        ExecutionContext jobContext = chunkContext.getStepContext()
            .getStepExecution()
            .getJobExecution()
            .getExecutionContext();

	// ExecutionContext 에 user.name 추가 
        jobContext.put("user.name", name);

        System.out.printf(HELLO_WORLD, name);
        return RepeatStatus.FINISHED;
    }
}

    // step에 추가할 listener
    @Bean
    public StepExecutionListener promotionListener()
    {
        ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
		
        // listener에서 name 키 승격
        listener.setKeys(new String[] {"name"});
        return listener;
    }

Step

step은 job의 구성 요소로 자체적인 입력, 출력, 처리를 가진다. state machine으로 생각하면 된다. 트랜잭션은 step 내에서 이뤄진다.

Tasklet 기반 Step

Tasklet 기반 Step을 만드는데는 두가지 유형이 있다.

1. MethodInvokingTaskletAdapter
사용자가 작성한 코드를 Tasklet Step 처럼 실행하는 방식
일반 POJO를 Step으로 활용 가능

2. Tasklet 인터페이스 구현
지금까지 위에서 사용한 방식이다. 이 때 Tasklet 인터페이스는 함수형 인터페이스이므로 람다식으로 구현할 수도 있다.

Tasklet 구현체의 처리가 완료되면 RepeatStatus 객체를 리턴해야 한다. (이게 바로 state machine으로 생각되는 지점!)

public enum RepeatStatus {
    CONTINUABLE(true),		// 어떤 조건이 충족될 때까지 반복 실행
    FINISHED(false);		// 성공 여부 관계 없이 tasklet 처리 완료 후 다음 처리
}

그 외 3가지 다른 Tasklet 구현체

1. CallableTaskletAdapter
Tasklet이 Step이 실행되는 스레드와 별개로 새로운 스레드에서 실행된다. 하지만 병렬로 실행되지는 않는다.
Callable 객체가 RepeatStatus를 리턴하기 전까지는 해당 Step이 완료된 것으로 간주되지 않는다. 그래서 다음 Step이 실행될 수 없다.

    @Bean
    public Callable<RepeatStatus> callableObject()
    {
        return () -> {
            System.out.println("This was executed in another thread");
            
	    // Callable 객체가 RepeatStatus를 리턴해야 tasklet 완료로 간주
            return RepeatStatus.FINISHED;
        };
    }

    @Bean
    public CallableTaskletAdapter tasklet()
    {
        CallableTaskletAdapter callableTaskletAdapter = new CallableTaskletAdapter();

        callableTaskletAdapter.setCallable(callableObject());

        return callableTaskletAdapter;
    }

2. MethodInvokingTaskletAdapter
MethodInvokingTaskletAdapter를 사용하면 기존의 다른 클래스의 메서드를 Step 내의 Tasklet으로 사용 가능하다.

    @StepScope
    @Bean
    public MethodInvokingTaskletAdapter methodInvokingTasklet(@Value("#{jobParameters['message']}") String message)
    {
        MethodInvokingTaskletAdapter methodInvokingTaskletAdapter = new MethodInvokingTaskletAdapter();

        methodInvokingTaskletAdapter.setTargetObject(service());
        methodInvokingTaskletAdapter.setTargetMethod("serviceMethod");
        methodInvokingTaskletAdapter.setArguments(new String[] {message});

        return methodInvokingTaskletAdapter;
    }

    // 내가 생성한 클래스(일반 POJO)
    @Bean
    public CustomService service()
    {
    	// 이 때 CustomService 클래스의 serviceMethod가 message 파라미터를 이용하지 않으면 에러 발생
        return new CustomService();
    }

3. SystemCommandTasklet
시스템 명령을 사용할 때 사용하며 해당 명령은 비동기로 실행된다.

    @Bean
    public SystemCommandTasklet systemCommandTasklet()
    {
        SystemCommandTasklet systemCommandTasklet = new SystemCommandTasklet();

        systemCommandTasklet.setCommand("rm -rf /tmp.txt");
        systemCommandTasklet.setTimeout(5000);
        // 시스템 명령이 비정상 종료될때 스레드를 강제 종료할지 여부 설정
        systemCommandTasklet.setInterruptOnCancel(true);

        return systemCommandTasklet;
    }

추가로 다양한 기능을 소개한다.

    @Bean
    public SystemCommandTasklet systemCommandTasklet()
    {
        SystemCommandTasklet systemCommandTasklet = new SystemCommandTasklet();

        systemCommandTasklet.setCommand("rm -rf /tmp.txt");
        systemCommandTasklet.setTimeout(5000);
        systemCommandTasklet.setInterruptOnCancel(true);

	// working directory 설정
        systemCommandTasklet.setWorkingDirectory("/Users/we/spring-batch");

	// ExitCode 설정
        systemCommandTasklet.setSystemProcessExitCodeMapper(touchCodeMapper());
        systemCommandTasklet.setTerminationCheckInterval(5000);
        // Lock이 걸리지 않도록 비동기 executor 설정
        systemCommandTasklet.setTaskExecutor(new SimpleAsyncTaskExecutor());
        // 환경 변수 설정
        systemCommandTasklet.setEnvironmentParams(new String[] {
            "JAVA_HOME=/java",
            "BATCH_HOME=/Users/batch"
        });

        return systemCommandTasklet;
    }

    @Bean
    public SimpleSystemProcessExitCodeMapper touchCodeMapper()
    {	
    	// 종료 상태에 따라 ExitStatus.COMPLETED, ExitStatus.FAILED 리턴
        return new SimpleSystemProcessExitCodeMapper();
    }

Chunk 기반 Step

    @Bean
    public Step step1()
    {
        return this.stepBuilderFactory.get("step1")
            // chunk size 는 10
            // 10개의 레코드를 읽고 처리할 때까지 쓰기 작업 하지 않음
            .<String, String>chunk(10)
            .reader(itemReader(null))
            .writer(itemWriter(null))
            .build();
    }

    @Bean
    @StepScope
    public FlatFileItemReader<String> itemReader(@Value("#{jobParameters['inputFile']}") Resource inputFile)
    {
        return new FlatFileItemReaderBuilder<String>()
            .name("itemReader")
            .resource(inputFile)
            .lineMapper(new PassThroughLineMapper())
            .build();
    }

    @Bean
    @StepScope
    public FlatFileItemWriter<String> itemWriter(@Value("#{jobParameters['outputFile']}") Resource outputFile)
    {
        return new FlatFileItemWriterBuilder<String>()
            .name("itemWriter")
            .resource(outputFile)
            .lineAggregator(new PassThroughLineAggregator<>())
            .build();
    }

크기가 동일하지 않은 Chunk를 처리할 때

이 때는 Chunk 크기를 하드 코딩을 할 수 없다. 이 때 사용하는 게 SimpleCompletionPolicy, TimeoutTerminationPolicy이다.

    @Bean
    public Step chunkStep()
    {
        return this.stepBuilderFactory.get("chunkStep")
            .<String, String>chunk(completionPolicy())
            .reader(itemReader())
            .writer(itemWriter())
            .build();
    }

    @Bean
    public ListItemReader<String> itemReader()
    {
        List<String> items = new ArrayList<>(10000);

        for (int i = 0; i < 10000; i++)
        {
            items.add(UUID.randomUUID().toString());
        }

        return new ListItemReader<>(items);
    }

    @Bean
    public ItemWriter<String> itemWriter()
    {
        return items -> {
            for (String item: items)
            {
                System.out.println(">> current item = " + item);
            }
        };
    }

    @Bean
    public CompletionPolicy completionPolicy()
    {
        CompositeCompletionPolicy policy = new CompositeCompletionPolicy();

        policy.setPolicies(
            new CompletionPolicy[]
                {
                	// 청크 처리 시간이 넘을 경우 안전하게 빠져나오게 해줌
                    // timeout 발생시 해당 청크 완료된 것으로 간주
                    new TimeoutTerminationPolicy(3),
                    // 100개의 레코드씩 끊어서 작업
                    new SimpleCompletionPolicy(100)
                }
        );

        return policy;
    }

TimeoutTerminationPolicy로 인해 커밋 개수가 10000/100 = 100 개가 아닌 105개 인 것을 확인할 수 있다.

Step Listener

Job Listener와 비슷하게 AfterStep, BeforeStep, AfterChunk, BeforeChunk를 이용해서 이벤트 처리를 할 수 있다.

Step Flow

조건 로직

on 메서드를 이용해서 step의 ExitStatus에 따라 어떤 일을 수행할지 결정할 수 있다.

@Bean
public Job job()
{
	return this.jobBuilderFactory.get("conditionalJob")
    	.start(firstStep())
        // 실패시 failureStep
        .on("FAILED").to(failureStep())
        // FAILED 외의 경우엔 successStep
        .from(firstStep()).on("*").to(sucessStep())
        .end()
        .build();
}

* : 0개 이상의 문자와 일치
? : 1개의 문자와 일치

decider를 따로 둬서 이에 따라 다음 step을 결정하게끔 할 수도 있다.

public class RandomDecider implements JobExecutionDecider
{
    private Random random = new Random();
    
    @Override
    public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution)
    {
        if (random.nextBoolean())
            return new FlowExecutionStatus(FlowExecutionStatus.COMPLETED.getName());
        else
            return new FlowExecutionStatus(FlowExecutionStatus.FAILED.getName());
    }
}

잡 종료하기

Completed : 성공적으로 종료. 동일 파라미터로 다시 실행 불가
Failed : 성공적으로 종료되지 않음. 동일 파라미터로 실행 가능
Stopped : 다시 시작 가능. 중단된 위치부터 잡을 다시 시작할 수 있음

@Bean
public Job job()
{
	return this.jobBuilderFactory.get("conditionalJob")
    	.start(firstStep())
        
        .on("FAILED").end()		// 스텝이 리턴한 상태에 상관없이 COMPLETED 저장
        
        .on("FAILED").fail()		// 실패하면 FAILED. 다시 실행 가능
        
        .on("FAILED").stopAndRestart(sucessStep())		// FAILED로 종료되지만 재실행 시 successStep부터 실행
        
        .from(firstStep()).on("*").to(successStep())
        .build();
}

flow 외부화하기

Step 의 정의를 추출해서 재사용 가능한 컴포넌트로 만들 수 있다. 여기엔 두가지 방법이 있다.

1. 스텝의 시퀀스를 독자적인 flow로 만드는 방법

 @Bean
 Public Flow preprocessingFlow()
 {
 	return new FlowBuilder<Flow>("preProcessingFlow").start(loadFileStep())
    .next(loadCustomerStep())
    .next(updateStartStep())
    .build();
 }

flow builder를 이용해서 flow를 생성하여 jobBuilder에게 넘긴다.

2. flow step을 사용하는 방법
위와 비슷하지만 flow를 step을 wrapping하고 해당 step을 job builder로 전달한다.

@Bean
public Step initializeBatch()
{
	return this.stepBuilderFactory.get("initializeBatch")
    	.flow(preprocessingFlow())
        .build();
}

1번 방법과 무슨 차이가 있을까? 바로 JobRepository에서 차이가 난다. 1번 방법을 사용하면 job에 step을 구성하는 것과 결과적으로 동일하다. 하지만 2번 방법은 flow가 담긴 step을 하나의 step처럼 기록한다. 이를 통해서 개별 step을 집계하지 않고도 flow의 영향을 전체적으로 모니터링 가능하다.

3. job 내에서 다른 job을 호출하는 방법
job을 step으로 wrapping하고 해당 step은 또 다른 job에서 호출된다.

@Bean
public Step initializeBatch()
{
	return this.stepBuilderFactory.get("initializeBatch")
    	.job(preprocessingJob())
        .parameterExtractor(new DefaultJobParametersExtractor())
        .build();
}

하지만 위의 방법은 job과의 의존성을 높여서 개별 job의 관리가 어려워지는 문제점이 생길 수 있으므로 피하는 것이 좋다.

문지은

백엔드 개발자입니다.

이전 포스트

JWT 토큰에 대한 보다 더 심화된 고찰

다음 포스트