배치 처리 완전 정복 — Spring Batch 개념부터 성능 최적화까지

dependencies { implementation 'org.springframework.boot:spring-boot-starter-batch' implementation 'org.springframework.boot:spring-boot-starter-data-jpa' implementation 'org.springframework.boot:spring-boot-starter-jdbc' runtimeOnly 'com.mysql:mysql-connector-j' testImplementation 'org.springframework.batch:spring-batch-test' }

spring: batch: job: enabled: false # 앱 시작 시 자동 실행 비활성화 jdbc: initialize-schema: always # 메타 테이블 자동 생성 (운영은 never) datasource: url: jdbc:mysql://localhost:3306/batchdb username: root password: password driver-class-name: com.mysql.cj.jdbc.Driver jpa: hibernate: ddl-auto: update show-sql: false

@Configuration @EnableBatchProcessing @RequiredArgsConstructor public class BatchConfig { private final JobRepository jobRepository; private final PlatformTransactionManager transactionManager; private final DataSource dataSource; // Job, Step 빈 등록은 이 클래스 또는 별도 JobConfig 클래스에서 관리 }

@Entity @Table(name = "orders") @Getter @NoArgsConstructor public class Order { @Id @GeneratedValue(strategy = GenerationType.IDENTITY) private Long id; private Long userId; private Long productId; private Integer amount; // 주문 금액 private String status; // COMPLETED, CANCELLED private LocalDate orderDate; } @Entity @Table(name = "settlement") @Getter @Setter @NoArgsConstructor public class Settlement { @Id @GeneratedValue(strategy = GenerationType.IDENTITY) private Long id; private Long userId; private Integer totalAmount; // 정산 금액 private LocalDate settlementDate; private LocalDateTime createdAt; }

@Configuration @RequiredArgsConstructor public class OrderSettlementJobConfig { private final JobRepository jobRepository; private final PlatformTransactionManager transactionManager; private final EntityManagerFactory entityManagerFactory; private final SettlementRepository settlementRepository; private static final int CHUNK_SIZE = 1000; @Bean public Job orderSettlementJob() { return new JobBuilder("orderSettlementJob", jobRepository) .start(orderSettlementStep()) .build(); } @Bean public Step orderSettlementStep() { return new StepBuilder("orderSettlementStep", jobRepository) .<Order, Settlement>chunk(CHUNK_SIZE, transactionManager) .reader(orderItemReader(null)) .processor(orderItemProcessor()) .writer(settlementItemWriter()) .faultTolerant() .skip(Exception.class) .skipLimit(10) // 10건까지 스킵 허용 .retry(DataAccessException.class) .retryLimit(3) // DB 오류 시 3회 재시도 .build(); } // JobParameter로 날짜 받기 @Bean @StepScope public JpaPagingItemReader<Order> orderItemReader( @Value("#{jobParameters['targetDate']}") String targetDate) { LocalDate date = LocalDate.parse(targetDate); Map<String, Object> params = new HashMap<>(); params.put("orderDate", date); params.put("status", "COMPLETED"); return new JpaPagingItemReaderBuilder<Order>() .name("orderItemReader") .entityManagerFactory(entityManagerFactory) .pageSize(CHUNK_SIZE) .queryString( "SELECT o FROM Order o " + "WHERE o.orderDate = :orderDate " + "AND o.status = :status " + "ORDER BY o.id ASC" ) .parameterValues(params) .build(); } @Bean public ItemProcessor<Order, Settlement> orderItemProcessor() { return order -> { // null 반환 시 해당 아이템은 Writer에 전달되지 않음 (필터링) if (order.getAmount() <= 0) return null; Settlement settlement = new Settlement(); settlement.setUserId(order.getUserId()); settlement.setTotalAmount(order.getAmount()); settlement.setSettlementDate(order.getOrderDate()); settlement.setCreatedAt(LocalDateTime.now()); return settlement; }; } @Bean public ItemWriter<Settlement> settlementItemWriter() { return items -> settlementRepository.saveAll(items); } }

@Component @RequiredArgsConstructor @Slf4j public class SettlementScheduler { private final JobLauncher jobLauncher; private final Job orderSettlementJob; // 매일 새벽 1시 실행 @Scheduled(cron = "0 0 1 * * *") public void runSettlementJob() { String targetDate = LocalDate.now().minusDays(1).toString(); try { JobParameters params = new JobParametersBuilder() .addString("targetDate", targetDate) .addLong("timestamp", System.currentTimeMillis()) // 재실행 보장 .toJobParameters(); JobExecution execution = jobLauncher.run(orderSettlementJob, params); log.info("[Settlement] Job 완료 | status={} | date={}", execution.getStatus(), targetDate); } catch (Exception e) { log.error("[Settlement] Job 실행 실패 | date={}", targetDate, e); } } }

@Component @RequiredArgsConstructor @Slf4j public class CleanupTasklet implements Tasklet { private final OrderRepository orderRepository; @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { // 90일 이상 된 취소 주문 삭제 LocalDate cutoff = LocalDate.now().minusDays(90); int deleted = orderRepository.deleteByStatusAndOrderDateBefore("CANCELLED", cutoff); log.info("[Cleanup] 취소 주문 {}건 삭제 완료", deleted); // FINISHED: 한 번 실행 후 종료 // CONTINUABLE: 계속 반복 실행 (페이징 처리 시 사용) return RepeatStatus.FINISHED; } } // Step 등록 @Bean public Step cleanupStep() { return new StepBuilder("cleanupStep", jobRepository) .tasklet(cleanupTasklet, transactionManager) .build(); }

@Bean public Job monthlyReportJob() { return new JobBuilder("monthlyReportJob", jobRepository) // Step 1: 원시 데이터 집계 .start(aggregateDataStep()) // Step 2: 집계 결과로 리포트 생성 .next(generateReportStep()) // Step 3: 생성된 리포트 이메일 발송 .next(sendReportStep()) // Step 3 성공 시 완료 .on("COMPLETED").end() // Step 3 실패 시 → 실패 알림 Step 실행 .from(sendReportStep()).on("FAILED").to(notifyFailureStep()) .end() .build(); }

@Bean public Step aggregateDataStep() { return new StepBuilder("aggregateDataStep", jobRepository) .tasklet((contribution, chunkContext) -> { long count = orderRepository.countByOrderDate(LocalDate.now().minusDays(1)); if (count == 0) { // 처리할 데이터 없음 → 커스텀 ExitStatus contribution.setExitStatus(new ExitStatus("NO_DATA")); } else { contribution.setExitStatus(ExitStatus.COMPLETED); } return RepeatStatus.FINISHED; }, transactionManager) .build(); } @Bean public Job conditionalJob() { return new JobBuilder("conditionalJob", jobRepository) .start(aggregateDataStep()) .on("NO_DATA").end() // 데이터 없으면 즉시 종료 .on("COMPLETED").to(generateReportStep()) // 있으면 리포트 생성 .end() .build(); }

// 일반적인 기준점 // - 단순 읽기/쓰기: 500 ~ 2000 // - 복잡한 변환 로직 포함: 100 ~ 500 // - 대용량 필드(CLOB/BLOB 포함): 50 ~ 200 private static final int CHUNK_SIZE = 1000; // 실제 튜닝은 처리 시간 측정 후 결정 // Step 실행 시간 로그: StepExecutionListener 활용 @Bean public StepExecutionListener stepTimingListener() { return new StepExecutionListenerSupport() { @Override public void beforeStep(StepExecution stepExecution) { log.info("[Step 시작] {}", stepExecution.getStepName()); } @Override public ExitStatus afterStep(StepExecution stepExecution) { log.info("[Step 완료] {} | 읽기={} | 쓰기={} | 스킵={} | 소요={}ms", stepExecution.getStepName(), stepExecution.getReadCount(), stepExecution.getWriteCount(), stepExecution.getSkipCount(), stepExecution.getEndTime().getTime() - stepExecution.getStartTime().getTime()); return stepExecution.getExitStatus(); } }; }

@Bean public Job parallelStepsJob() { // Flow 1: 사용자 정산 Flow userSettlementFlow = new FlowBuilder<Flow>("userSettlementFlow") .start(userSettlementStep()) .build(); // Flow 2: 상품 통계 Flow productStatsFlow = new FlowBuilder<Flow>("productStatsFlow") .start(productStatsStep()) .build(); // 두 Flow를 병렬 실행 Flow parallelFlow = new FlowBuilder<Flow>("parallelFlow") .split(new SimpleAsyncTaskExecutor()) // 병렬 실행 Executor .add(userSettlementFlow, productStatsFlow) .build(); return new JobBuilder("parallelStepsJob", jobRepository) .start(parallelFlow) .next(finalizeStep()) // 병렬 완료 후 최종 처리 .end() .build(); }

@Bean public Step multiThreadedStep() { return new StepBuilder("multiThreadedStep", jobRepository) .<Order, Settlement>chunk(CHUNK_SIZE, transactionManager) .reader(orderItemReader(null)) .processor(orderItemProcessor()) .writer(settlementItemWriter()) .taskExecutor(taskExecutor()) // 멀티스레드 적용 .throttleLimit(4) // 동시 스레드 수 제한 .build(); } @Bean public TaskExecutor taskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(4); executor.setMaxPoolSize(8); executor.setQueueCapacity(25); executor.setThreadNamePrefix("batch-thread-"); executor.initialize(); return executor; }

// thread-safe 래핑 @Bean @StepScope public SynchronizedItemStreamReader<Order> synchronizedOrderReader( @Value("#{jobParameters['targetDate']}") String targetDate) { return new SynchronizedItemStreamReaderBuilder<Order>() .delegate(orderItemReader(targetDate)) .build(); }

// Partitioner: 데이터를 어떻게 나눌지 정의 @Component public class OrderPartitioner implements Partitioner { private final OrderRepository orderRepository; @Override public Map<String, ExecutionContext> partition(int gridSize) { Long minId = orderRepository.findMinId(); Long maxId = orderRepository.findMaxId(); long range = (maxId - minId) / gridSize + 1; Map<String, ExecutionContext> result = new HashMap<>(); for (int i = 0; i < gridSize; i++) { long startId = minId + (range * i); long endId = startId + range - 1; ExecutionContext context = new ExecutionContext(); context.putLong("startId", startId); context.putLong("endId", Math.min(endId, maxId)); result.put("partition-" + i, context); } return result; } } // Manager Step: 파티션을 조율하는 상위 Step @Bean public Step partitionManagerStep() { return new StepBuilder("partitionManagerStep", jobRepository) .partitioner("workerStep", orderPartitioner) .step(workerStep()) // 각 파티션이 실행할 Step .gridSize(8) // 파티션 수 .taskExecutor(taskExecutor()) .build(); } // Worker Step: 실제 처리 @Bean public Step workerStep() { return new StepBuilder("workerStep", jobRepository) .<Order, Settlement>chunk(CHUNK_SIZE, transactionManager) .reader(partitionedOrderReader(null, null)) .processor(orderItemProcessor()) .writer(settlementItemWriter()) .build(); } // 파티션 파라미터로 범위 제한 @Bean @StepScope public JdbcPagingItemReader<Order> partitionedOrderReader( @Value("#{stepExecutionContext['startId']}") Long startId, @Value("#{stepExecutionContext['endId']}") Long endId) { Map<String, Object> params = new HashMap<>(); params.put("startId", startId); params.put("endId", endId); SqlPagingQueryProviderFactoryBean queryProvider = new SqlPagingQueryProviderFactoryBean(); queryProvider.setSelectClause("SELECT id, user_id, product_id, amount, order_date"); queryProvider.setFromClause("FROM orders"); queryProvider.setWhereClause("WHERE id BETWEEN :startId AND :endId AND status = 'COMPLETED'"); queryProvider.setSortKey("id"); try { return new JdbcPagingItemReaderBuilder<Order>() .name("partitionedOrderReader") .dataSource(dataSource) .queryProvider(queryProvider.getObject()) .parameterValues(params) .pageSize(CHUNK_SIZE) .rowMapper(new BeanPropertyRowMapper<>(Order.class)) .build(); } catch (Exception e) { throw new RuntimeException(e); } }

@Bean public JdbcBatchItemWriter<Settlement> jdbcBatchSettlementWriter() { return new JdbcBatchItemWriterBuilder<Settlement>() .dataSource(dataSource) .sql( "INSERT INTO settlement (user_id, total_amount, settlement_date, created_at) " + "VALUES (:userId, :totalAmount, :settlementDate, :createdAt)" ) .beanMapped() // Settlement 필드명과 :파라미터 자동 매핑 .build(); } // MySQL rewriteBatchedStatements=true 설정 시 실제 배치 Insert 활성화 // url: jdbc:mysql://host/db?rewriteBatchedStatements=true

@Bean public Step faultTolerantStep() { return new StepBuilder("faultTolerantStep", jobRepository) .<Order, Settlement>chunk(CHUNK_SIZE, transactionManager) .reader(orderItemReader(null)) .processor(orderItemProcessor()) .writer(settlementItemWriter()) .faultTolerant() // Skip: 특정 예외 발생 시 해당 아이템을 건너뜀 .skip(InvalidDataException.class) // 데이터 검증 실패 .skip(NullPointerException.class) .skipLimit(100) // 최대 100건까지만 스킵 (초과 시 Job 실패) // Retry: 일시적 오류 시 재시도 .retry(DataAccessException.class) // DB 커넥션 오류 등 .retryLimit(3) // 3회 재시도 // NoSkip: 절대 스킵하면 안 되는 예외 .noSkip(OutOfMemoryError.class) // 스킵된 아이템 로깅 .listener(skipListener()) .build(); } @Bean public SkipListener<Order, Settlement> skipListener() { return new SkipListenerSupport<>() { @Override public void onSkipInProcess(Order item, Throwable t) { log.warn("[Skip] 처리 중 스킵 | orderId={} | error={}", item.getId(), t.getMessage()); } @Override public void onSkipInWrite(Settlement item, Throwable t) { log.warn("[Skip] 저장 중 스킵 | userId={} | error={}", item.getUserId(), t.getMessage()); } }; }

@Component @RequiredArgsConstructor @Slf4j public class SettlementJobListener implements JobExecutionListener { private final SlackNotifier slackNotifier; @Override public void beforeJob(JobExecution jobExecution) { log.info("[Job 시작] {} | params={}", jobExecution.getJobInstance().getJobName(), jobExecution.getJobParameters()); } @Override public void afterJob(JobExecution jobExecution) { BatchStatus status = jobExecution.getStatus(); long readCount = jobExecution.getStepExecutions().stream() .mapToLong(StepExecution::getReadCount).sum(); long writeCount = jobExecution.getStepExecutions().stream() .mapToLong(StepExecution::getWriteCount).sum(); if (status == BatchStatus.COMPLETED) { log.info("[Job 완료] 읽기={} 쓰기={}", readCount, writeCount); } else if (status == BatchStatus.FAILED) { log.error("[Job 실패] status={}", status); // Slack 알림 발송 slackNotifier.sendAlert("정산 배치 실패!"); } } } // Job에 리스너 등록 @Bean public Job orderSettlementJob() { return new JobBuilder("orderSettlementJob", jobRepository) .listener(settlementJobListener) .start(orderSettlementStep()) .build(); }

@Bean @StepScope public FlatFileItemReader<OrderCsv> csvOrderReader( @Value("#{jobParameters['filePath']}") String filePath) { return new FlatFileItemReaderBuilder<OrderCsv>() .name("csvOrderReader") .resource(new FileSystemResource(filePath)) .linesToSkip(1) // 헤더 스킵 .delimited() .delimiter(",") .names("userId", "productId", "amount", "orderDate") .targetType(OrderCsv.class) .build(); }

// JobParameters에 날짜를 포함시켜 같은 날짜는 재실행 안 됨 // 단, FAILED 상태의 Job은 재시작 가능 (timestamp 없이) JobParameters params = new JobParametersBuilder() .addString("targetDate", "2026-04-28") // timestamp 제거 → 동일 파라미터로 완료된 Job은 재실행 불가 .toJobParameters();

@Bean public Job nonRestartableJob() { return new JobBuilder("nonRestartableJob", jobRepository) .preventRestart() // 실패해도 재시작 불가 .start(step1()) .build(); }

-- Spring Batch 핵심 메타 테이블 -- BATCH_JOB_INSTANCE : Job 인스턴스 목록 -- BATCH_JOB_EXECUTION : Job 실행 이력 (상태, 시작/종료 시간) -- BATCH_STEP_EXECUTION : Step 실행 이력 (읽기/쓰기/스킵 건수) -- BATCH_JOB_EXECUTION_PARAMS : Job 실행 시 파라미터 -- 실행 이력 조회 SELECT ji.JOB_NAME, je.START_TIME, je.END_TIME, je.STATUS, je.EXIT_CODE, TIMESTAMPDIFF(SECOND, je.START_TIME, je.END_TIME) AS duration_sec FROM BATCH_JOB_EXECUTION je JOIN BATCH_JOB_INSTANCE ji ON je.JOB_INSTANCE_ID = ji.JOB_INSTANCE_ID ORDER BY je.START_TIME DESC LIMIT 20; -- 오래된 이력 정리 (90일 이상) DELETE FROM BATCH_JOB_EXECUTION WHERE START_TIME < DATE_SUB(NOW(), INTERVAL 90 DAY) AND STATUS IN ('COMPLETED', 'FAILED');