데이터 삽입 속도 (Mysql vs Batch)

팀원이 더미 데이터를 CSV 파일로 생성해주고, 이 데이터를 어떻게 DB에 넣을지 고민했다.

1. SQL Script 생성

2. Mysql CSV으로 삽입

몇천 개의 데이터를 직접 쿼리로 작성하는 노가다를 하기는 싫었다. 하나의 테이블에 기본 몇천 건 이상의 데이터가 존재하는데, 다른 테이블에도 똑같은 행동을 한다고 생각하니 머리가 아팠다.

그래서 Mysql에 CSV 파일을 직접 넣을 수 있어서 이 방법을 택했다.

Mysql

삽입하는 방식도 두 가지로 생각했다.

터미널에 명령어를 입력하는 방법과 work bench로 GUI 환경에서 파일 경로를 입력하는 방법이다.

work bench에서 CSV 파일을 직접 넣어보았다.

5532개의 레코드를 처리하는데 무료 123.640s 라는 시간이 걸렸다. 클라이언트의 request에 대한 응답 시간이 저 정도가 아니라 단순히 데이터를 넣는 작업이니 실행시켜놓고 기다리면 상관없다고 생각은 했지만, 10만 건의 데이터를 넣는다고 생각하면 정말 오래 걸릴 것이다. 더 좋은 방법이 없을까 고민하다가 이 기회에 Spring Batch를 시도해 봤다.

Spring Batch

스프링 배치를 시도한 이유는 다음과 같다.

1. 자동화가 가능하다.

2. 읽기, 쓰기, 처리에 대한 대용량 데이터를 처리할 수 있다.

3. 도중에 오류가 발생해도 개발자의 설정으로 스킵 하거나 특정 단계를 다시 수행이 가능하다.

4. 트랜잭션을 지원한다.

• Mysql로 삽입하다가 중간에 에러가 발생하면 처음부터 다시 해야 된다. (데이터 정합성 같은)

이 외에도 정말 많은 이점들이 존재할 것이다.

구현

@Configuration()
@RequiredArgsConstructor
public class BatchConfig {
    
    private final JobRepository jobRepository;


    @Bean
    public Job perfumeFileReadAndWriteJob(PlatformTransactionManager transactionManager) {
        return new JobBuilder("job name", jobRepository)
                .incrementer(new RunIdIncrementer())
                .start(step)
                .build();
    }

    // step 정의
    ...
    // Writer 정의
    ...
    // Processor 정의 - 필수 요소는 아님
    ...
    // bean Item Reader 정의
    ...
}

프로젝트는 배포 자동화가 되있어서 변경이 있으면 알아서 배포한다.

이 방식은 Bean으로 등록해버리기 때문에 서버가 다시 구동되면 Job이 실행되고 같은 데이터가 계속 쌓이고 있었다.

어떻게 할지 고민하다가 Job의 실행도 API로 만들어 버려야겠다고 생각했다. 구현이 조금 달라졌다.

✅ Config

@Configuration()
@RequiredArgsConstructor
public class BatchConfig {

    private final JobRepository jobRepository;
    private final PerfumeBatchItem perfumeBatchItem;
    ...
    
    @Bean
    public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor(JobRegistry jobRegistry){
        JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor = new JobRegistryBeanPostProcessor();
        jobRegistryBeanPostProcessor.setJobRegistry(jobRegistry);
        return jobRegistryBeanPostProcessor;
    }

    ...
}

✅ Controller

@RestController
@RequiredArgsConstructor
public class RestBatchController {

    private final JobLauncher jobLauncher;
    private final JobRegistry jobRegistry;
    private final JobExplorer jobExplorer;

    @PostMapping("/batch/job")
    public ExitStatus runJobBrand(@RequestBody JobRequestDto request) throws Exception {
        Job job = jobRegistry.getJob(request.jobName());

        JobParameters jobParameters = new JobParametersBuilder(jobExplorer)
                .getNextJobParameters(job)
                .toJobParameters();

        return jobLauncher.run(job, jobParameters).getExitStatus();
    }
}

요청으로 Job 이름을 보내면 해당 Job을 호출 시킨다. 그리고 jobRegistry로 job을 실행시킨다.

JobRegistry는 컨텍스트 내에 있는 Job을 추적한다. @EnableBatchProcessing를 선언하면 기본적으로 제공하지만 버전이 바뀌고 사용할 수 없게되어 빈으로 등록한다. JobRegistryBeanPostProcessor는 BeanPostProcessor의 구현체로 생성된 Bean이 Job일 경우 JobRegistry에 등록해 준다.

워크 벤치에서는 120초 이상이 걸렸지만 Batch로 처리한 것은 10초도 안걸렸다.

이렇게 빠른 이유는 Chunk 덕분이 아닐까?

/**
 * Batch domain interface representing the configuration of a step. As with a {@link Job},
 * a {@link Step} is meant to explicitly represent the configuration of a step by a
 * developer but also the ability to execute the step.
 *
 * @author Dave Syer
 * @author Mahmoud Ben Hassine
 *
 */
public interface Step {

	/**
	 * The key to use when retrieving the batch step type.
	 */
	String STEP_TYPE_KEY = "batch.stepType";

	/**
	 * @return the name of this step.
	 */
	String getName();

	/**
	 * @return {@code true} if a step that is already marked as complete can be started
	 * again. Defaults to {@code false}.
	 */
	default boolean isAllowStartIfComplete() {
		return false;
	}

	/**
	 * @return the number of times a step can be (re)started for the same job instance.
	 * Defaults to {@code Integer.MAX_VALUE}
	 */
	default int getStartLimit() {
		return Integer.MAX_VALUE;
	}

	/**
	 * Process the step and assign progress and status meta information to the
	 * {@link StepExecution} provided. The {@link Step} is responsible for setting the
	 * meta information and also saving it, if required by the implementation.<br>
	 *
	 * It is not safe to reuse an instance of {@link Step} to process multiple concurrent
	 * executions.
	 * @param stepExecution an entity representing the step to be executed.
	 * @throws JobInterruptedException if the step is interrupted externally.
	 */
	void execute(StepExecution stepExecution) throws JobInterruptedException;

}

Job은 내가 정의한 Step을 처리하는데, Step에서 chunk를 설정해줬다.

public Step fileReadStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
        return new StepBuilder("step name", jobRepository)
                .<Object, Object>chunk(10, transactionManager)
                ..
                .build();
}

Chunk는 데이터를 한 번에 읽어서 처리 후 DB에 저장한다. 지정한 Chunk 사이즈가 10이고 데이터가 100개라면 총 10개의 Chunk로 데이터가 나뉜다. 각 Chunk는 SimpleAsyncTaskExecutor에 의해서 별도의 스레드에서 실행된다. 즉, (데이터 / Chunk 사이즈)로 병렬 처리 되면서 속도가 엄청 빨라지는 것 같다.

또한 Chunk는 트랜잭션을 관리한다. Chunk의 크기가 10이면 한 번에 10개의 데이터를 처리하고 DB에 write한다.

중간에 에러가 발생한다면 문제가 있는 Chunk를 rollback 할 수 있다. (무결성)

Chunk 프로세싱