티스토리

채마스의 개발창고

검색하기

블로그 홈

채마스의 개발창고

hyunwook.dev/m

채마스 님의 블로그입니다.

구독자: 0

방명록 방문하기

주요 글 목록

대용량 작업 분산처리하기(With Spring Batch) 개요 만약 1억 건 이상의 데이터를 가공해야 하는 작업이 있다면 어떻게 처리하면 좋을까? 병렬처리를 한다고 해도 단일 서버에서 1억 건을 처리하는 것은 무리가 있다. 그렇기 때문에 여러 서버에서 분산처리하는 것이 바람직하다. 이번 글에서는 대용량 작업을 Redis Streams를 활용해서 분산처리할 수 있는 방법에 대해서 정리해보려고 한다. 필수 사전 지식 Scaling and Parallel Processing(With Spring Batch) Redis Streams 먼저 Redis Streams에서 사용할 명령어를 간단하게 정리해 보자. Redis Streams Consumer Group 생성 Redis Streams에서 컨슈머 그룹을 생성하는 명령어는 아래와 같다. XGROUP CREATE [st.. 공감수 0 댓글수 0 2023. 6. 4.
Scaling and Parallel Processing(With Spring Batch) 개요 Spring Batch는 대용량 데이터를 처리하는 데 최적화되어 있는 기능들을 제공하는 프레임워크다. 나는 실제로 Spring Batch를 사용해서 대용량 데이터 처리를 효율적으로 처리했다. 이번 글에서는 Spring Batch를 사용해서 대용량 데이터를 처리하는 방법에 대해서 정리해 보려고 한다. 필수 사전 지식 Spring Batch Architecture 왜 Spring Batch가 대용량 데이터 처리에 유리한가? Spring Batch가 대용량 데이터 처리에 유리한 이유는 크게 3가지다. Chunk-oriented Processing Spring Batch는 대량의 데이터를 Chunk 단위로 나누어 처리한다. 그렇기 때문에 메모리를 효율적으로 사용할 수 있다. 작업 실패 시, 실패한 Chun.. 공감수 0 댓글수 0 2023. 5. 20.
ExecutionContext ExecutionContext 란? Job에서 사용하는 데이터를 보관하는 보관소이다. Job처리를 통해서 언제나 참조 가능한 데이터를 보존하고, 추가 및 갱신도 가능하다. step처리의 결과 (결과값 등)를 다음 Step에 전달하는 것은 불가능하다. JobExecutionContext에 저장해서 어떤 step에서도 참조 가능하게 할 수 있다. Step 내 처리에서 데이터를 참조, 추가, 갱신이 자유롭게 가능하다. JobExecutonContext와 달리, 다른 step에서 참조가 불가능하다. 예를들어, chunk 처리에서 도중에 에러 종료로 끝난 경우 -> 어디까지 데이터를 읽어 들어드렸는지를 stepExecutionContext에 보관한다. -> 실패했던 곳에서부터 처리를 Step 을 재처리 할 수 있.. 공감수 0 댓글수 0 2022. 4. 2.
JobParameterValidator JobParameterValidator JobParameterValidator를 사용해서 Job 파라미터를 validation 처리할 수 있다. 만약 Job 파라미터로 날짜값이 전달 될때, 날짜 형식의 파라미터 맞는지 validation 처리해 보는 코드를 구현해 보자 먼저 아래와 같이 날짜 형식이 맞는지 검사하는 Validator를 구현한다. @AllArgsConstructor public class LocalDateParameterValidator implements JobParametersValidator { private String parameterName; @Override public void validate(JobParameters parameters) throws JobParameters.. 공감수 0 댓글수 0 2022. 4. 2.
Job, Step 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.. 공감수 0 댓글수 0 2022. 4. 2.
Spring Batch 메타 테이블 메타테이블이란? -> 배치 실행을 위한 메타 데이터가 저장되는 테이블이다. 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.. 공감수 0 댓글수 0 2022. 4. 2.
성능 평가 Async Step vs Multi-Thread Step vs Partition Step vs Parallel Step 성능 평가 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.. 공감수 0 댓글수 0 2022. 2. 27.
JobExecutionDecider 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.. 공감수 0 댓글수 0 2022. 2. 27.
Skip, Retry Skip 예외처리 step 수행 중 발생한 특정 Exception과 에러 횟수 설정으로 예외처리 설정 만약 skip(NotFoundNameException.class), skipLimit(3)으로 설정한다면 NotFoundNameException이 3번 발생할 때까지는 에러를 skip 한다. 4번째에서 Job과 Step의 상태가 실패로 끝나며, 배치가 중지된다. 여기서 에러 발생 전까지는 모두 정상처리된 상태로 남는다. 그렇기 때문에 오류가 난 다음부터 다시 시작해야 된다. -> 안 그러면 중복이 발생함 ex> 10개의 chunk 중 1~9까지 정삭으로 작동하고 10에서 오류가 났다면 -> 배치 재 실행 시 10번째 chunk부터 처리하도록 배치를 구현해야 된다. .faultTolerant() .skip.. 공감수 0 댓글수 0 2022. 2. 27.
StepListener 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() 메.. 공감수 0 댓글수 0 2022. 2. 27.
JobExecutionListener JobExecutionListener 스프링 배치에서 전 처리, 후 처리를 하기 위해 사용된다. 2가지의 구현 방식이 있다. interface 구현 @Annotation 정의 JobExecutionListener를 interface 형식으로 구현하면 아래와 같다. public static class SavePersonJobExecutionListener implements JobExecutionListener{ // job 실행전에 호출 @Override public void beforeJob(JobExecution jobExecution) { log.info("beforeJob"); } // Job 실행 후에 호 @Override public void afterJob(JobExecution jobExe.. 공감수 0 댓글수 0 2022. 2. 27.
ItemWriter ItemWriter interface ItemWriter는 마지막으로 배치 처리 대상 데이터를 어떻게 처리할 지 결정한다. DB에 데이터를 저장하거나, API로 서버에 요청하거나, 파일에 데이터를 write하거나 최종적으로 할 작업을 진행한다. Step에서 ItemWriter는 필수이다. csvItemWriter private ItemWriter csvItemWriter() throws Exception { BeanWrapperFieldExtractor fieldExtractor = new BeanWrapperFieldExtractor(); fieldExtractor.setNames(new String[] {"id", "name", "age", "address"}); DelimitedLineAggrega.. 공감수 0 댓글수 0 2022. 2. 27.
ItemProcessor ItemProcessor ItemWriter에 데이터를 보낼지 말지를 결정함 ItemReader에서 null을 리턴하면 chunk처리가 끝났다는 의미 ItemProcessor에서 null을 리턴하면 ItemWriter에서 해당 item을 처리하지 않겠다는 의미 ItemProcessor는 필수는 아니지만, 책임 분리를 위해 사용하는 것이 좋다. 구현은 아래와 같이 할 수 있다. @Service public class PlayerSalaryService { public PlayerSalaryDto calcSalary(PlayerDto player) { int salary = (Year.now().getValue() - player.getBirthYear()) * 1000000; return PlayerSal.. 공감수 0 댓글수 0 2022. 2. 27.
ItemReader Custom ItemReader ItemReader 를 커스터마이징 아래와 같은 코드 방식으로 커스터마이징 할 수 있다. public class CustomItemReader implements ItemReader { private final List items; public CustomItemReader(List items) { this.items = new ArrayList(items); } @Override public Object read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException { if(!items.isEmpty()) { return items.remove(0); } ret.. 공감수 0 댓글수 0 2022. 2. 27.
Tasklet, Chunk 비교 Tasklet, Chunk 비교 Tasklet을 사용한 Task 기반 처리 Chunk보다 코딩하기 쉽다. 배치 처리 과정이 비교적 쉬운 경우 사용한다. 대량 처리를 하는 경우에는 더 복잡하다. 하나의 큰 덩어리를 여러 덩어리로 나누어 처리하기에는 부적합하다. Chunk를 사용한 chunk(덩어리) 기반 처리 ItemReader, ItemProcessor, ItemWriter의 관계를 이해할 필요가 있다. 대량 처리를 하는 경우 Tasklet 보다 비교적 쉽게 구현할 수 있다. Ex> 10000개의 데이터를 1000개씩 10개의 덩어리로 나누어 실행 @JobScope, @StepScope @Scope는 어떤 시점에 bean을 생성/소멸 시킬지 bean의 lifecycle을 설정할 수 있다. @JobScop.. 공감수 0 댓글수 0 2022. 2. 27.
Spring Batch batch란? 큰 단위의 작업을 일괄 처리 대용량 처리량이 많고 비 실시간성 처리에 사용 대용량 데이터 계산, 정산, 통계, 데이터베이스, 변환 등 컴퓨터 자원을 최대로 활용 컴퓨터 자원 사용이 낮은 시간대에 배치를 처리하거나 배치만 처리하기 위해 사용자가 사용하지 않는 또다른 컴퓨터 자원을 사용 사용자 상호작용으로 실행되기 보단, 스케줄러와 같은 시스템에 의해 실행되는 대상 crontab, jenkins 등을 이용해서 주기적으로 실행할 수 있다. 배치 프로그램은 보이진 않지만 늘 존재한다. 서비스를 운영하는 관점에서 주기적으로 작업을 처리하려면 배치 프로그램을 사용해야 한다. batch 가 필요한 상황 필요한 데이터를 모아서 처리해야 할 때 ex. 월별 거래 명세서 생성 일부러 지연시켜 처리할 때 ex.. 공감수 0 댓글수 0 2022. 2. 27.

문의안내

티스토리
로그인
고객센터

티스토리는 카카오에서 사랑을 담아 만듭니다.