[Spring Batch] Step 간 데이터 공유 (StepExecutionContext)

Spring Batch

 

 

chunk 방식의 Step을 사용하여 batch 소스를 작성하던 중 Step 간 데이터 공유가 필요한 상황이 생겨 해당 문제를 처리한 방법을 작성한다.

 

// 배치 버전
spring-batch-core:5.1.0
spring-boot-starter-batch:3.2.0

 

 

Spring 공식 문서에서는 Step ExecutionContext에 데이터를 저장 후 꺼내어 사용하라고 권장하고 있고 내가 작업한 순서는 다음과 같다.

 

1. StepExecutionListener 구현 및 StepExecution 세팅

나는 Step ExecutionContext 객체를 사용하기 위해 StepExecution 객체가 필요했고 해당 객체를 가져오기 위해 스텝 작업 전후를 확인할 수 있는 StepExecutionListener 인터페이스를 구현하였다.

 

그리고 StepExecutionListener 구현 시 beforeStep 메서드에서 인스턴스 변수인 stepExecution을 세팅하였고 해당 리스너를 StepBuilder에 추가하였다.

(beforeStep 메서드는 리스너를 추가한 스텝이 실행되기 전에 실행되는 메서드이다)

 

참고로 두 번째 스탭에서는 해당 리스너를 추가하지 않았다.

추가를 하게 된다면 StepExecution이 초기화되어 저장한 데이터를 꺼내어 사용할 수 없기 때문이다.

@Slf4j
@Configuration
public class FileDbToDbDeleteJobConfig extends AbstractBatchJobConfig {

    @Setter
    private StepExecution stepExecution;
    
    private static final String STEP_KEY = "KEY";

    @StepScope
    @Bean
    public StepExecutionListener fileDbToDbDeleteStepExecutionListener() {
        return new StepExecutionListener() {
            @Override
            public void beforeStep(StepExecution stepExecution) {
                setStepExecution(stepExecution);
            }
        };
    }
    
    @Bean
    @JobScope
    public Step fileDbToDbDeleteStep() {
        return new StepBuilder(JOB_NAME + "Step", jobRepository)
                .<FileInDto, FileOutDto>chunk(CHUNK_SIZE, transactionManager)
                .reader(fileDbToDbDeleteReader())
                .processor(fileDbToDbDeleteProcessor())
                .writer(fileDbToDbDeleteWriter())
                .listener(fileDbToDbDeleteStepExecutionListener()) // listener 추가
                .build();
    }
}

 

 

2. Step ExecutionContext 객체에 데이터 저장

인스턴스 변수인 stepExecution 객체에서 ExecutionContext를 가져온 후 put 메서드를 사용하여 데이터를 삽입한다.

 

ExecutionContext는 ConcurrentHashMap으로 구성되어 있는데 내가 개발하고 있는 시스템은 멀티 스레드 환경이 아니라서 HashMap이어도 문제는 없다.

ExecutionContext HashMap

 

 

나의 경우 최대 1,000개의 UUID 문자열 데이터가 필요하여 첫 번째 Step ItemReader에서 데이터를 조회한 후 ItemProcessor에서 인스턴수 int 타입의 변수인 stepKeyCount를 증가시키며 데이터를 삽입하였다.

 

첫 번째 Step

    @Bean
    @StepScope
    public ItemProcessor<FileInDto, FileOutDto> fileDbToDbDeleteProcessor() {
        return fileInDto -> {
            FileOutDto fileOutDto = FileOutDto.builder()
                    .fileKey(fileInDto.getFileKey())
                    .fileUuid(fileInDto.getFileUuid())
                    .build();

            if (this.stepExecution == null) {
                throw new NullPointerException("StepExecution is null");
            }

            ExecutionContext executionContext = this.stepExecution.getExecutionContext();
            
            // 데이터 삽입 - 테스트용 소스
            for (int i = 1; i <= 1000; i++) {
                executionContext.put(STEP_KEY + ++stepKeyCount, "18542389-2206-11ef-a912-3d9ce8f5400b");
            }

            String saveFileNm = fileInDto.getSaveFileNm();
            String saveFileRute = fileInDto.getSaveFileRute();

            if (saveFileRute == null || saveFileNm == null) {
                return fileOutDto;
            }

            boolean isDelete = FileUtil.deleteFile(saveFileRute, saveFileNm);
            log.info("$$$$$$$$$$$ isDelete = {}", isDelete);

            return fileOutDto;
        };
    }

 

 

3. Step ExecutionContext 객체에 저장한 데이터 추출

두 번째 Step

첫 번째 Step의 ItemProcessor에서 저장한 데이터를 꺼내왔다.

@Bean
@StepScope
public MyBatisCursorItemReader<FileDtlInDto> fileDbToFileDeleteReader() {

    if (this.stepExecution == null) {
        throw new NullPointerException("StepExecution is null");
    }

    ExecutionContext executionContext = this.stepExecution.getExecutionContext();

    // 데이터 가져오기 - 테스트 소스
    for (int i = 1; i <= stepKeyCount; i++) {
        log.debug("$$$$$$$$$$ data = {}", executionContext.get(STEP_KEY + i, String.class));
    }

    return new MyBatisCursorItemReaderBuilder<FileDtlInDto>()
            .sqlSessionFactory(sqlSessionFactory)
            .queryId("FileMapper.getFileIn")
            .saveState(true)
            .build();
}

 

 

UUID 1000개의 데이터에 대해서 저장 가능한 것을 확인하였다.

결과 로그

데이터 저장 시 주의 사항

배치 메타 테이블 중 BATCH_STEP_EXECUTION_CONTEXT 테이블은 아래와 같이 구성되어 있는데

Step ExecutionContext 객체에 데이터 저장 시 SHORT_CONTEXT에 값이 저장되어 사이즈가 큰 데이터를 저장하면 오류가 발생한다고 한다.

create table BATCH_STEP_EXECUTION_CONTEXT
(
    STEP_EXECUTION_ID  NUMBER(19)          not null
        primary key
        constraint STEP_EXECUTION_CONTEXT_FK
            references BATCH_STEP_EXECUTION,
    SHORT_CONTEXT      VARCHAR2(2500 char) not null,
    SERIALIZED_CONTEXT CLOB
)

 

그러나 실제로 확인해 보니 SERIALIZED_CONTEXT 컬럼에 값이 업데이트되고 SHORT_CONTEXT에는 트림 처리가 된 SERIALIZED_CONTEXT 값이 저장되기 때문에 너무 큰 값만 아니면 괜찮은 것 같다.

(버전의 차이가 있는 듯하다)

 

아래 SQL은 STEP이 실행되면서 발생한 쿼리인데 SHORT_CONTEXT 컬럼에는 '...'으로 생략되었다는 표시로 데이터가 저장된다.

 

UPDATE BATCH_STEP_EXECUTION_CONTEXT
SET SHORT_CONTEXT = '생략 AfgADdAAPZmls ...', SERIALIZED_CONTEXT = 'rO0ABXNyABFqYXZhLnV0aWwuS 생략'
WHERE STEP_EXECUTION_ID = 345

 

 

전체 소스 (Step 분기 및 Exit Message 설정 포함)

@Slf4j
@Configuration
public class FileDbToDbDeleteJobConfig extends AbstractBatchJobConfig {

    private static final int CHUNK_SIZE = 10;

    private static final String STEP_KEY = "KEY";

    private int stepKeyCount = 0;

    private static final ExitStatus exitStatus = new ExitStatus(ExitStatus.COMPLETED.getExitCode(), "There is no selected data.");

    public static final String JOB_NAME = "fileDbToDbDeleteJob";
    
    @Setter
    private StepExecution stepExecution;

    public FileDbToDbDeleteJobConfig(SqlSessionFactory sqlSessionFactory,
                                     JobRepository jobRepository,
                                     PlatformTransactionManager transactionManager) {
        super(sqlSessionFactory, jobRepository, transactionManager);
    }

    /**
     * 조회된 데이터가 없는 경우 Exit Message를 설정한다
     * @return fileDbToDbDeleteJobExecutionListener
     */
    @JobScope
    @Bean
    public JobExecutionListener fileDbToDbDeleteJobExecutionListener() {
        return new JobExecutionListener() {
            @Override
            public void afterJob(JobExecution jobExecution) {
                List<StepExecution> stepExecutions = (List<StepExecution>) jobExecution.getStepExecutions();

                for (StepExecution stepExecution : stepExecutions) {
                    if ("fileDbToDbDeleteJobTerminateStep".equals(stepExecution.getStepName())) {
                        if (ExitStatus.COMPLETED.getExitCode().equals(stepExecution.getExitStatus().getExitCode())) {
                            jobExecution.setExitStatus(exitStatus);
                            return;
                        }
                    }
                }
            }
        };
    }

    /**
     * step 간 데이터 공유를 위해 stepExecution 객체를 세팅한다
     * @return fileDbToDbDeleteStepExecutionListener
     */
    @StepScope
    @Bean
    public StepExecutionListener fileDbToDbDeleteStepExecutionListener() {
        return new StepExecutionListener() {
            @Override
            public void beforeStep(StepExecution stepExecution) {
                setStepExecution(stepExecution);
            }
        };
    }

    /**
     * 조회된 데이터가 없는 경우 로그 남기고 배치를 종료한다
     * @param fileDbToDbDeleteStep 파일 디테일 테이블 삭제 Step
     * @param fileDbToDbDeleteFileStep 파일 마스터 테이블 삭제 Step
     * @param fileDbToDbDeleteTerminateStep Terminate Step 
     * @return JobBuilder
     */
    @Bean(JOB_NAME)
    public Job fileDbToDbDeleteJob(Step fileDbToDbDeleteStep, Step fileDbToDbDeleteFileStep, Step fileDbToDbDeleteTerminateStep) {
        return new JobBuilder(JOB_NAME, jobRepository)
                .start(fileDbToDbDeleteStep)
                .next(fileDbToDbDeleteJobExecutionDecider())
                .from(fileDbToDbDeleteJobExecutionDecider())
                    .on(FlowExecutionStatus.COMPLETED.getName())
                    .to(fileDbToDbDeleteFileStep)
                .from(fileDbToDbDeleteJobExecutionDecider())
                    .on(FlowExecutionStatus.STOPPED.getName())
                    .to(fileDbToDbDeleteTerminateStep)
                    .end()
                .listener(fileDbToDbDeleteJobExecutionListener())
                .build();
    }

    /**
     * 조회된 데이터가 있는지 확인한다
     * @return fileDbToDbDeleteJobExecutionDecider
     */
    @Bean
    public JobExecutionDecider fileDbToDbDeleteJobExecutionDecider() {
        return (JobExecution jobExecution, StepExecution stepExecution) -> {
            assert stepExecution != null;
            return stepExecution.getReadCount() == 0L ? new FlowExecutionStatus(FlowExecutionStatus.STOPPED.getName()) : new FlowExecutionStatus(FlowExecutionStatus.COMPLETED.getName());
        };
    }

    @Bean
    @JobScope
    public Step fileDbToDbDeleteStep() {
        return new StepBuilder(JOB_NAME + "Step", jobRepository)
                .<FileInDto, FileOutDto>chunk(CHUNK_SIZE, transactionManager)
                .reader(fileDbToDbDeleteReader())
                .processor(fileDbToDbDeleteProcessor())
                .writer(fileDbToDbDeleteWriter())
                .listener(fileDbToDbDeleteStepExecutionListener())
                .build();
    }

    @Bean
    @StepScope
    public MyBatisCursorItemReader<FileInDto> fileDbToDbDeleteReader() {
        return new MyBatisCursorItemReaderBuilder<FileInDto>()
                .sqlSessionFactory(sqlSessionFactory)
                .queryId("FileMapper.getFileInList")
                .saveState(true)
                .build();
    }

    @Bean
    @StepScope
    public ItemProcessor<FileInDto, FileOutDto> fileDbToDbDeleteProcessor() {
        return fileInDto -> {
            FileOutDto fileOutDto = FileOutDto.builder()
                    .fileKey(fileInDto.getFileKey())
                    .fileUuid(fileInDto.getFileUuid())
                    .build();

            if (this.stepExecution == null) {
                throw new NullPointerException("StepExecution is null");
            }

            ExecutionContext executionContext = this.stepExecution.getExecutionContext();
            executionContext.put(STEP_KEY + ++stepKeyCount, fileOutDto.getFileKey());

            String saveFileNm = fileInDto.getSaveFileNm();
            String saveFileRute = fileInDto.getSaveFileRute();

            if (saveFileRute == null || saveFileNm == null) {
                return fileDtlOutDto;
            }

            boolean isDelete = FileUtil.deleteFile(saveFileRute, saveFileNm);
            log.info("$$$$$$$$$$$ isDelete = {}", isDelete);

            return fileOutDto;
        };
    }

    @Bean
    @StepScope
    public MyBatisBatchItemWriter<FileOutDto> fileDbToDbDeleteWriter() {
        return new MyBatisBatchItemWriterBuilder<FileOutDto>()
                .sqlSessionFactory(sqlSessionFactory)
                .statementId("FileMapper.deleteFileList")
                .assertUpdates(false)
                .build();
    }

    @Bean
    @JobScope
    public Step fileDbToDbDeleteFileStep() {
        return new StepBuilder(JOB_NAME + "FileStep", jobRepository)
                .<FileInDto, FileInDto>chunk(CHUNK_SIZE, transactionManager)
                .reader(fileDbToFileDeleteReader())
                .writer(fileDbToFileDeleteWriter())
                .build();
    }

    @Bean
    @StepScope
    public MyBatisCursorItemReader<FileInDto> fileDbToFileDeleteReader() {

        if (this.stepExecution == null) {
            throw new NullPointerException("StepExecution is null");
        }

        ExecutionContext executionContext = this.stepExecution.getExecutionContext();

        List<String> uuidList = new ArrayList<>();

        for (int i = 1; i <= stepKeyCount; i++) {
            uuidList.add(executionContext.get(STEP_KEY + i, String.class));
        }

        return new MyBatisCursorItemReaderBuilder<FileDtlInDto>()
                .sqlSessionFactory(sqlSessionFactory)
                .queryId("FileMapper.getFileIn")
                .parameterValues(Map.of("uuidList", uuidList))
                .saveState(true)
                .build();
    }

    @Bean
    @StepScope
    public MyBatisBatchItemWriter<FileInDto> fileDbToFileDeleteWriter() {
        return new MyBatisBatchItemWriterBuilder<FileDtlInDto>()
                .sqlSessionFactory(sqlSessionFactory)
                .statementId("FileMapper.deleteFile")
                .assertUpdates(false)
                .build();
    }

    @Bean
    @JobScope
    public Step fileDbToDbDeleteTerminateStep() {
        return new StepBuilder(JOB_NAME + "TerminateStep", jobRepository)
                .tasklet(fileDbToDbDeleteTerminateTasklet(), transactionManager)
                .build();
    }

    @Bean
    @StepScope
    public Tasklet fileDbToDbDeleteTerminateTasklet() {
        return (contribution, chunkContext) -> {
            contribution.setExitStatus(exitStatus);
            return RepeatStatus.FINISHED;
        };
    }
}

 

 

 

- Spring Batch 표지 이미지 출처

https://djlife.tistory.com/31

 

- 참고 레퍼런스

https://velog.io/@gongmeda/Spring-Batch%EC%97%90%EC%84%9C-Step%EA%B0%84-%EB%8D%B0%EC%9D%B4%ED%84%B0-%EA%B3%B5%EC%9C%A0%ED%95%98%EB%8A%94-%EB%B0%A9%EB%B2%95 
https://fvor001.tistory.com/107