Spring Batch ‐ Tasklet vs Chunk - thought-corner/backend-roadmap GitHub Wiki

• Spring Batch가 제공하는 Tasklet 인터페이스의 execute() 메서드에 우리가 원하는 로직을 구현하고, 이 구현체를 Spring Batch에 넘기기만 하면 된다.
• 그 이후의 실행과 흐름 관리는 Spring Batch가 알아서 처리한다.

@FunctionalInterface
public interface Tasklet {
    @Nullable
    RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception;
}

• RepeatStatus.FINISHED : Step의 처리가 성공이든 실패든 상관없이 해당 Step이 완료되었음을 의미한다. 더 이상 반복할 필요 없이 다음 스텝으로 넘어가며, 배치 잡은 차근차근 진행된다.
• RepeatStatus.CONTINUABLE : Tasklet의 execute() 메서드가 추가로 더 실행되어야 함을 Spring Batch Step에 알리는 신호다. Step의 종료는 보류되고, 필요한 만큼 execute() 메서드가 반복 호출된다.

• execute() 내부에서 while문을 사용한다면 N만 건째 처리 중 예외가 발생했을 때, 이미 처리한 (N-1)만 건의 데이터도 모두 롤백되어 하나도 정리되지 않은 상태로 돌아간다.
• RepeatStatus.CONTINUABLE로 반복한다면 매 만 건 처리마다 트랜잭션이 커밋되므로, 예외가 발생하더라도 (N-1)만 건의 데이터는 이미 안전하게 정리된 상태로 남는다.

• 결국 RepeatStatus를 반환해 execute()를 반복 실행하도록 하는 이유는 거대한 하나의 트랜잭션 대신 작은 트랜잭션들로 나누어 안전하게 처리하기 위해서다.

• ResourcelessTransactionManager는 no-op(아무것도 하지 않는) 방식으로 동작하는 PlatformTransactionManager 구현체로 이를 사용하면 불필요한 DB 트랜잭션 처리를 생략할 수 있다.
• ResourcelessTransactionManager를 Step에 적용하고자 한다면 다음과 같이 ResourcelessTransactionManager 인스턴스를 tasklet() 메서드에 전달해주면 된다.

❗주의사항 : PlatformTransactionManager 빈을 직접 정의할 땐 주의가 필요

• Spring Batch는 내부적으로 Job과 Step의 상태와 같은 메타데이터를 DB를 통해 관리한다.
• 이 때도 트랜잭션이 사용되는데, 별도의 구성 변경 없이 PlatformTransactionManager를 빈으로 정의할 경우 Step의 비즈니스 로직 처리를 위한 트랜잭션과 메타데이터 관리를 위한 트랜잭션이 서로 다른 성격임에도 불구하고 같은 PlatformTransactionManager 빈을 사용하게 되어 의도치 않은 문제가 발생할 수 있다.

• 단순 작업에 적합 : 태스크릿 지향 처리는 알림 발송, 파일 복사, 오래된 데이터 삭제 등 단순 작업을 처리하는 Step 유형이다.
• Tasklet 인터페이스 구현 : Tasklet 인터페이스를 구현해 필요한 로직을 작성한 뒤, 이를 StepBuilder.tasklet() 메서드에 전달해 Step을 구성한다.
• RepeatStatus로 실행 제어 : Tasklet.execute() 메서드는 RepeatStatus를 반환하며, 이를 통해 실행 반복 여부를 결정할 수 있다.
• 트랜잭션 지원 : Spring Batch는 Tasklet.execute() 메서드 실행 전후로 트랜잭션을 시작하고 커밋하여, 데이터베이스의 일관성과 원자성을 보장한다.

• 청크(Chunk)는 데이터를 일정 단위로 쪼갠 덩어리를 말한다.
• Spring Batch에서 데이터 기반 처리 방식을 청크 지향 처리라고 부르는 이유는 읽고, 처리하고, 쓰는 작업을 일정 크기로 나눈 데이터 덩어리(청크)를 대상으로 하기 때문이다.
• 100만 건의 데이터를 처리해야 한다고 가정해보자. Spring Batch는 100만 건 전체를 한 번에 읽고 처리하고 쓰지 않는다. 대신, 100개씩 쪼개서 읽고, 처리하고, 저장한다. 이렇게 나뉜 100개의 묶음이 바로 청크다.

• 메모리 절약 : 100만 건을 한 번에 메모리에 올리는 건 문제가 된다. DB에서 데이터를 불러오는 순간, 메모리 과부하로 시스템에 장애가 발생할 수 있다.
• 가벼운 트랜잭션 : 트랜잭션은 작업의 성공 또는 실패를 하나의 단위로 묶는 것이다. 하지만 100만 건을 하나의 트랜잭션으로 처리하다가 작업 중간에 오류가 발생하면 100만 건이 전부 롤백된다.

public interface ItemReader<T> {
    T read() throws Exception, 
    UnexpectedInputException, 
    ParseException, 
    NonTransientResourceException;
}

• read() 메서드는 아이템을 하나씩 반환한다. 여기서 아이템이란 파일의 한 줄 또는 데이터베이스의 한 행(row)에 해당하는 데이터 하나를 의미한다.
• ItemReader가 null을 반환하는 것이 청크 지향 처리 Step의 종료 시점이라는 의미가 된다.

public interface ItemProcessor<I, O> {
    O process(I item) throws Exception;
}

• 데이터 가공 : 입력 데이터(I)를 원하는 형태(O)로 변환한다.
• 필터링 : process() 메서드가 null을 반환하면 해당 입력 데이터는 처리 흐름에서 제외된다. 다시 말해 ItemWriter로 전달되지 않는다. 유효하지 않은 데이터나 처리할 필요가 없는 데이터를 걸러낼 때 사용된다.
• 데이터 검증 : 입력 데이터의 유효성을 검사한다. 필터링과 달리 조건에 맞지 않는 데이터를 만나면 예외를 발생시킨다.
• 필수 아님 : ItemProcessor는 생략 가능하다.

public interface ItemWriter<T> {
    void write(Chunk<? extends T> chunk) throws Exception;
}

• 한 덩어리씩 쓴다 : ItemWriter는 데이터를 한 건씩 쓰지 않는다. Chunk 단위로 묶어서 한 번에 데이터를 쓴다. write() 메서드의 파라미터 타입이 Chunk이다.

1. 데이터 읽기(ItemReader)

• ItemReader는 데이터 소스에서 하나씩 데이터를 읽어온다.
• read() 메서드가 호출될 때마다 데이터를 순차적으로 반환하며, 청크 크기만큼 데이터를 읽어야 끝난다.

2. 데이터 깎기(ItemProcessor)

• ItemProcessor는 ItemReader가 읽어온 청크의 각 아이템 하나씩을 처리한다.
• process() 메서드는 청크 전체를 입력받지 않는다. 청크의 각 아이템을 하나씩 받아서 처리한다.

3. 데이터 쓰기(ItemWriter)

• ItemProcessor의 처리까지 완료되었다면 이제 청크를 실제로 쓸 차레가 된다.
• ItemReader / ItemProcessor가 각각의 아이템을 하나씩 처리하는 것과 달리, ItemWriter는 청크 전체를 한 번에 입력받는다.

• 청크 사이즈가 클 때 : 그만큼 메모리에 많은 데이터를 한 번에 로드하게 된다. 트랜잭션의 경계가 커지므로, 문제 발생시 롤백되는 데이터의 양도 많아진다.
• 청크 사이즈가 작을 때 : 트랜잭션의 경계가 작아져서 문제 발생시 롤백되는 데이터가 최소화된다. 대신 그만큼 읽기/쓰기 I/O가 자주 발생하게 된다.

• JobParameters는 배치 작업에 전달되는 입력 값이다. 이 값은 배치가 어떤 조건에서 어떤 데이터를 다룰지를 결정하는 데 핵심적인 역할을 한다.

• --spring.batch.job.name으로 전달가능하다.
• JobParameters 기본 표기법은 다음과 같다. ⭢ parameterName=parameterValue,parameterType,identificationFlag
  • parameterName : 배치 Job에서 파라미터를 찾을 때 사용할 key 값이다. 이 이름으로 Job 내에서 파라미터에 접근할 수 있다.
  • parameterValue : 파라미터의 실제 값
  • parameterType : 파라미터의 타입(java.lang.String, java.lang.Integer와 같은 fully qualified name 사용). 이 파라미터 타입을 명시하지 않을 경우 Spring Batch는 해당 파라미터를 String 타입으로 가정한다.
  • identificationFlag : Spring Batch에게 해당 파라미터가 JobInstance 식별(identification)에 사용될 파라미터인지 여부를 전달하는 값으로 true이면 식별에 사용된다는 의미이다. 이 플래그는 생략 가능하며 생략할 경우 true로 설정된다.

• 파라미터 값에 쉼표를 사용하고 싶다면 Spring Batch 5부터 지원해주는 JSON 기반의 파라미터 표기법을 사용하면 된다.

// 의존성 추가
dependencies {
    // 기존 다른 의존성들...
    implementation 'org.springframework.boot:spring-boot-starter-json'
}

// JsonJobParametersConverter를 빈으로 등록
@Bean
public JobParametersConverter jobParametersConverter() {
    return new JsonJobParametersConverter();
}

infiltrationTargets='{"value": "시청_서버실,안산_데이터센터", "type": "java.lang.String"}'

• 프로그래밍 방식으로 JobParameters를 생성/전달하려면 JobParametersBuilder라는 컴포넌트가 필요하다.

JobParameters jobParameters = new JobParametersBuilder()
    .addJobParameter("inputFilePath", "/data/input/users.csv", String.class)
    .toJobParameters();

jobLauncher.run(dataProcessingJob, jobParameters);

• JobParameters에 직접 접근하려면 먼저 JobParameters가 어디에 저장되고 관리되는지를 이해해야 한다.
• Spring Batch에서는 JobExecution이라는 클래스가 Job의 실행 정보를 쥐고 있다. JobParameters도 이 안에 있다.
• StepExecution은 스텝 실행에 관한 정보를 담고 있는 객체이면서 내부적으로 부모 Job의 JobExecution을 참조하고 있기에 StepExecution을 통해 파라미터를 가져올 수 있다.

• JobParametersValidator를 사용하면 잘못된 파라미터가 들어오는 순간 즉시 차단할 수 있다.

public interface JobParametersValidator {
    void validate(@Nullable JobParameters parameters) throws JobParametersInvalidException;
}