[Spring Batch] 대용량 배치 처리
spring batch, Job, Step, ItemReader, ItemProcessor, ItemWriter, chunk processing
정의
Spring Batch는 대용량 데이터의 배치 처리 (ETL, 일괄 통계, 마이그레이션 등)를 위한 프레임워크. Job → Step → ItemReader → ItemProcessor → ItemWriter 청크 모델. 실패 재시도, 트랜잭션, 메타데이터(이력) 영속화.
설정
implementation("org.springframework.boot:spring-boot-starter-batch")@SpringBootApplication
@EnableBatchProcessing // 5.0+ optional (자동)
public class App { }배치 메타데이터 테이블 자동 생성:
BATCH_JOB_INSTANCE,BATCH_JOB_EXECUTIONBATCH_STEP_EXECUTION,BATCH_JOB_EXECUTION_PARAMS
Job + Step (Chunk)
@Configuration
public class ImportUsersJobConfig {
@Bean
public Job importUsersJob(JobRepository repo, Step importStep) {
return new JobBuilder("importUsersJob", repo)
.start(importStep)
.build();
}
@Bean
public Step importStep(
JobRepository repo,
PlatformTransactionManager tx,
ItemReader<UserCsv> reader,
ItemProcessor<UserCsv, User> processor,
ItemWriter<User> writer
) {
return new StepBuilder("importStep", repo)
.<UserCsv, User>chunk(1000, tx) // 1000개 단위 트랜잭션
.reader(reader)
.processor(processor)
.writer(writer)
.faultTolerant()
.retryLimit(3).retry(IOException.class)
.skipLimit(10).skip(ParseException.class)
.build();
}
}ItemReader
CSV 파일
@Bean
@StepScope // job parameter 사용 위해
public FlatFileItemReader<UserCsv> reader(@Value("#{jobParameters['filename']}") String filename) {
return new FlatFileItemReaderBuilder<UserCsv>()
.name("userReader")
.resource(new FileSystemResource(filename))
.delimited()
.names("email", "name", "age")
.targetType(UserCsv.class)
.linesToSkip(1) // 헤더
.build();
}JDBC
@Bean
public JdbcCursorItemReader<UserCsv> reader(DataSource ds) {
return new JdbcCursorItemReaderBuilder<UserCsv>()
.name("userReader")
.dataSource(ds)
.sql("SELECT email, name, age FROM legacy_users WHERE migrated = false")
.rowMapper(new BeanPropertyRowMapper<>(UserCsv.class))
.build();
}
// Paged
@Bean
public JdbcPagingItemReader<UserCsv> pagedReader(DataSource ds) {
return new JdbcPagingItemReaderBuilder<UserCsv>()
.name("pagedReader")
.dataSource(ds)
.pageSize(1000)
.selectClause("SELECT email, name, age")
.fromClause("FROM legacy_users")
.sortKeys(Map.of("id", Order.ASCENDING))
.rowMapper(new BeanPropertyRowMapper<>(UserCsv.class))
.build();
}PagedReader는 큰 결과셋에 더 안전 (cursor 유지 X).
Custom
@Bean
public ItemReader<UserCsv> apiReader() {
return new ItemReader<>() {
private Iterator<UserCsv> iterator;
@Override
public UserCsv read() {
if (iterator == null) {
iterator = fetchAllFromApi().iterator();
}
return iterator.hasNext() ? iterator.next() : null; // null = 끝
}
};
}ItemProcessor
@Bean
public ItemProcessor<UserCsv, User> processor() {
return csv -> {
if (csv.email() == null || !csv.email().contains("@")) {
return null; // skip
}
return new User(csv.email().toLowerCase(), csv.name(), csv.age());
};
}null 반환 시 해당 item skip (writer로 안 감).
Composite Processor
@Bean
public ItemProcessor<UserCsv, User> processor(
ItemProcessor<UserCsv, UserCsv> validator,
ItemProcessor<UserCsv, User> converter
) {
return new CompositeItemProcessorBuilder<UserCsv, User>()
.delegates(List.of(validator, converter))
.build();
}ItemWriter
JDBC
@Bean
public JdbcBatchItemWriter<User> writer(DataSource ds) {
return new JdbcBatchItemWriterBuilder<User>()
.dataSource(ds)
.sql("INSERT INTO users (email, name, age) VALUES (:email, :name, :age)")
.beanMapped()
.build();
}JPA
@Bean
public JpaItemWriter<User> writer(EntityManagerFactory emf) {
JpaItemWriter<User> writer = new JpaItemWriter<>();
writer.setEntityManagerFactory(emf);
return writer;
}Custom
@Bean
public ItemWriter<User> writer(UserRepository userRepository) {
return users -> {
userRepository.saveAll(users.getItems());
};
}실행
@Service
public class JobRunner {
private final JobLauncher jobLauncher;
private final Job importUsersJob;
public void run(String filename) throws Exception {
JobParameters params = new JobParametersBuilder()
.addString("filename", filename)
.addLong("timestamp", System.currentTimeMillis()) // 매번 다른 instance
.toJobParameters();
jobLauncher.run(importUsersJob, params);
}
}CLI:
java -jar app.jar --spring.batch.job.name=importUsersJob filename=data.csv스케줄링은 @Scheduled 또는 Quartz.
다중 Step / Flow
@Bean
public Job complexJob(JobRepository repo, Step step1, Step step2, Step step3) {
return new JobBuilder("complexJob", repo)
.start(step1)
.next(step2)
.next(step3)
.build();
}
// 조건부 분기
@Bean
public Job conditionalJob(JobRepository repo, Step validate, Step process, Step rollback) {
return new JobBuilder("conditionalJob", repo)
.start(validate)
.on("FAILED").to(rollback)
.from(validate).on("*").to(process)
.end()
.build();
}병렬 처리
Multi-threaded Step
@Bean
public Step parallelStep(...) {
return new StepBuilder("parallel", repo)
.<In, Out>chunk(100, tx)
.reader(reader)
.processor(processor)
.writer(writer)
.taskExecutor(new SimpleAsyncTaskExecutor("worker"))
.throttleLimit(10) // 10 스레드 동시
.build();
}reader는 thread-safe해야 함.
Partitioning
@Bean
public Step partitionedStep(Step slaveStep) {
return new StepBuilder("partitioned", repo)
.partitioner("slaveStep", new ColumnRangePartitioner())
.step(slaveStep)
.gridSize(10)
.taskExecutor(...)
.build();
}데이터를 N개로 분할 → 각 partition을 별 스레드/노드에서 처리.
재시도 / Skip
.faultTolerant()
.retry(NetworkException.class).retryLimit(3)
.skip(ParseException.class).skipLimit(100)
.listener(new SkipListener<...>() {
public void onSkipInRead(Throwable t) {
log.warn("Skipped read", t);
}
public void onSkipInProcess(Object item, Throwable t) {
log.warn("Skipped processing {}", item, t);
}
public void onSkipInWrite(Object item, Throwable t) {
log.warn("Skipped write {}", item, t);
}
})Job 모니터링
BATCH_JOB_EXECUTION테이블 직접 조회- Spring Batch Admin UI (deprecated, 별도 dashboard 필요)
- Micrometer 메트릭 (자동)
JobExplorerAPI로 코드에서 조회
@Autowired
JobExplorer jobExplorer;
JobExecution last = jobExplorer.getJobExecutions(jobInstance).iterator().next();
last.getStatus(); // COMPLETED, FAILED, STOPPED
last.getStepExecutions();자주 보는 패턴
일일 CSV 가져오기
@Component
public class DailyImport {
@Scheduled(cron = "0 0 2 * * *")
public void run() {
Path file = Paths.get("/data/users-" + LocalDate.now() + ".csv");
jobLauncher.run(importJob, new JobParametersBuilder()
.addString("filename", file.toString())
.toJobParameters());
}
}Database 마이그레이션
레거시 DB → 새 스키마. JdbcReader → JpaWriter.
Report 생성
DB 집계 → 결과를 CSV/Excel 작성.
@Bean
public FlatFileItemWriter<Report> writer() {
return new FlatFileItemWriterBuilder<Report>()
.name("reportWriter")
.resource(new FileSystemResource("/reports/sales-" + LocalDate.now() + ".csv"))
.delimited()
.names("region", "month", "total")
.build();
}함정
1. JobParameters 유일성
같은 parameters로 두 번 실행 → 동일 instance, 이전 결과 재사용. 새 instance 원하면 timestamp 같은 unique 파라미터 추가.
2. chunk 크기
너무 작으면 트랜잭션 오버헤드, 너무 크면 메모리/롤백 비용. 1000~10000이 일반적. 측정 후 결정.
3. reader thread-safety
FlatFileItemReader는 기본 not thread-safe. multi-threaded step 사용 시 SynchronizedItemStreamReader로 감싸거나 partitioning.
4. Lazy initialization
@Bean
@StepScope
public ItemReader<X> reader(@Value("#{jobParameters['x']}") String x) { ... }@StepScope 안 붙이면 jobParameters resolved 실패.
5. 메타 테이블 권한
배치 메타 테이블이 DB에 쓰기 권한 필요. 읽기 전용 DB로는 실행 불가.
Spring Batch vs Alternatives
| 도구 | 특징 |
|---|---|
| Spring Batch | 표준, 풍부 |
| Apache Airflow | DAG 기반, 다언어 |
| Apache Spark | 매우 큰 데이터 |
| AWS Glue / GCP Dataflow | 매니지드 |
| Custom script | 작은 작업 |
대용량 분산은 Spark. 일반 enterprise 배치는 Spring Batch.
💬 댓글