채마스의 개발창고

개요 만약 1억 건 이상의 데이터를 가공해야 하는 작업이 있다면 어떻게 처리하면 좋을까? 병렬처리를 한다고 해도 단일 서버에서 1억 건을 처리하는 것은 무리가 있다. 그렇기 때문에 여러 서버에서 분산처리하는 것이 바람직하다. 이번 글에서는 대용량 작업을 Redis Streams를 활용해서 분산처리할 수 있는 방법에 대해서 정리해보려고 한다. 필수 사전 지식 Scaling and Parallel Processing(With Spring Batch) Redis Streams 먼저 Redis Streams에서 사용할 명령어를 간단하게 정리해 보자. Redis Streams Consumer Group 생성 Redis Streams에서 컨슈머 그룹을 생성하는 명령어는 아래와 같다. XGROUP CREATE [st..

개요 Spring Batch는 대용량 데이터를 처리하는 데 최적화되어 있는 기능들을 제공하는 프레임워크다. 나는 실제로 Spring Batch를 사용해서 대용량 데이터 처리를 효율적으로 처리했다. 이번 글에서는 Spring Batch를 사용해서 대용량 데이터를 처리하는 방법에 대해서 정리해 보려고 한다. 필수 사전 지식 Spring Batch Architecture 왜 Spring Batch가 대용량 데이터 처리에 유리한가? Spring Batch가 대용량 데이터 처리에 유리한 이유는 크게 3가지다. Chunk-oriented Processing Spring Batch는 대량의 데이터를 Chunk 단위로 나누어 처리한다. 그렇기 때문에 메모리를 효율적으로 사용할 수 있다. 작업 실패 시, 실패한 Chun..

ExecutionContext 란? Job에서 사용하는 데이터를 보관하는 보관소이다. Job처리를 통해서 언제나 참조 가능한 데이터를 보존하고, 추가 및 갱신도 가능하다. step처리의 결과 (결과값 등)를 다음 Step에 전달하는 것은 불가능하다. JobExecutionContext에 저장해서 어떤 step에서도 참조 가능하게 할 수 있다. Step 내 처리에서 데이터를 참조, 추가, 갱신이 자유롭게 가능하다. JobExecutonContext와 달리, 다른 step에서 참조가 불가능하다. 예를들어, chunk 처리에서 도중에 에러 종료로 끝난 경우 -> 어디까지 데이터를 읽어 들어드렸는지를 stepExecutionContext에 보관한다. -> 실패했던 곳에서부터 처리를 Step 을 재처리 할 수 있..

JobParameterValidator JobParameterValidator를 사용해서 Job 파라미터를 validation 처리할 수 있다. 만약 Job 파라미터로 날짜값이 전달 될때, 날짜 형식의 파라미터 맞는지 validation 처리해 보는 코드를 구현해 보자 먼저 아래와 같이 날짜 형식이 맞는지 검사하는 Validator를 구현한다. @AllArgsConstructor public class LocalDateParameterValidator implements JobParametersValidator { private String parameterName; @Override public void validate(JobParameters parameters) throws JobParameters..

Job 전체 배치 프로세스를 캡슐화한 도메인이다. Step의 순서를 정의한다. JobParameters를 받는다. @Bean public Job helloJob() { return jobBuilderFactory.get("helloJob") .incrementer(new RunIdIncrementer()) .start(this.helloStep()) // job 실행 시 최초로 실행될 step 설정 .build(); } job을 하나 생성한다. -> job은 batch의 실행단위를 말한다. job의 실행단위를 구분할 수 있는 incrementer -> RunIdIncrementer는 job이 실행될 때마다 Id를 자동으로 생성해 준다. job의 name을 "helloJob"으로 지정했는데 name은 sp..

메타테이블이란? -> 배치 실행을 위한 메타 데이터가 저장되는 테이블이다. Spring Batch Framework가 실행 시 meta data 테이블들을 사용하므로 초기 설정이 필요하다. Spring Batch Framework에 속하는 부분이므로 수정하지 않고 조회만 한다. Job의 이력, 파라미터등 실행 결과를 조회할 수 있다. 배치 결과에 대해서 로그, 별도의 실행 이력을 남기는 경우가 대부분이므로 조회할 일이 많지 않다. BATCH_JOB_INSTANCE Job이 실행되며 생성되는 최상위 계층의 테이블이다. job_name과 job_key를 기준으로 하나의 row가 생성된다. -> 중복은 있을 수 없다. job_key는 BATCH_JOB_EXECUTION_PARAMS에 저장되는 Parameter..

성능 평가 Async Step vs Multi-Thread Step vs Partition Step vs Parallel Step Async Step single thread 기반과 async Processing 기반을 비교하면 아래와 같다. single thread async thread ItemProcessor와 ItemWriter를 Async로 실행한다. java.util.concurrent에서 제공하는 Future를 기반으로 asynchronous를 실행한다. Async를 사용하기 위해 spring-batch-integration 디펜던시를 추가해 줘야한다. ItemProcessor는 아래와 같이 구현할 수 있다. private AsyncItemProcessor itemProcessor() { I..

JobExecutionDecider public class JobParametersDecide implements JobExecutionDecider { public static final FlowExecutionStatus CONTINUE = new FlowExecutionStatus("CONTINUE"); private final String key; public JobParametersDecide(String key) { this.key = key; } @Override public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) { String value = jobExecution.getJobPar..

Skip 예외처리 step 수행 중 발생한 특정 Exception과 에러 횟수 설정으로 예외처리 설정 만약 skip(NotFoundNameException.class), skipLimit(3)으로 설정한다면 NotFoundNameException이 3번 발생할 때까지는 에러를 skip 한다. 4번째에서 Job과 Step의 상태가 실패로 끝나며, 배치가 중지된다. 여기서 에러 발생 전까지는 모두 정상처리된 상태로 남는다. 그렇기 때문에 오류가 난 다음부터 다시 시작해야 된다. -> 안 그러면 중복이 발생함 ex> 10개의 chunk 중 1~9까지 정삭으로 작동하고 10에서 오류가 났다면 -> 배치 재 실행 시 10번째 chunk부터 처리하도록 배치를 구현해야 된다. .faultTolerant() .skip..

StepListener 위에서 구현한 StepExecutionListener 또한 StepListener를 상속받은 Listener이다. StepListener를 상속받는 Listener의 종류로는 아래와 같다. StepExecutionListener SkipListener onSkipInRead : @OnSkipInRead ItemReader에서 Skip이 발생한 경우 호출 onSkipInWrite : @OnSkipInWrite ItemWriter에서 Skip이 발생한 경우 호출 onSkipProcess : @OnSkipInProcess ItemProcessor에서 Skip이 발생한 경우 호출 ItemReadListener beforeRead : @BeforeRead ItemReader.read() 메..