Split View: Spring Boot Batch 완전 가이드: Job, Step, Chunk 처리와 실전 패턴

Job
└── Step 1
│   └── Chunk (ChunkSize: 100)
│       ├── ItemReader   (읽기)
│       ├── ItemProcessor (변환/필터)
│       └── ItemWriter   (쓰기)
└── Step 2
    └── Tasklet (단일 작업)

JobLauncher ──→ Job (실행 요청)
     │
     ↓
JobRepository (실행 이력 저장/조회)
     │
     ↓
JobExplorer (읽기 전용 이력 조회)

-- 주요 메타데이터 테이블
BATCH_JOB_INSTANCE    -- Job 인스턴스 정보
BATCH_JOB_EXECUTION   -- Job 실행 정보 (상태, 시작/종료 시간)
BATCH_JOB_PARAMS      -- Job 파라미터
BATCH_STEP_EXECUTION  -- Step 실행 정보
BATCH_STEP_EXECUTION_CONTEXT -- Step 컨텍스트 데이터
BATCH_JOB_EXECUTION_CONTEXT  -- Job 컨텍스트 데이터

<!-- Spring Boot Batch -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-batch</artifactId>
</dependency>

<!-- 배치 메타 테이블용 DB (예: PostgreSQL) -->
<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
</dependency>

<!-- 테스트 -->
<dependency>
    <groupId>org.springframework.batch</groupId>
    <artifactId>spring-batch-test</artifactId>
    <scope>test</scope>
</dependency>

spring:
  batch:
    job:
      enabled: false # 애플리케이션 시작 시 자동 실행 방지
    jdbc:
      initialize-schema: always # 메타 테이블 자동 생성
  datasource:
    url: jdbc:postgresql://localhost:5432/batchdb
    username: batchuser
    password: batchpass

logging:
  level:
    org.springframework.batch: DEBUG

@Configuration
@EnableBatchProcessing
public class UserMigrationJobConfig {

    // Job 정의
    @Bean
    public Job userMigrationJob(JobRepository jobRepository,
                                 Step migrationStep) {
        return new JobBuilder("userMigrationJob", jobRepository)
                .start(migrationStep)
                .listener(jobExecutionListener())
                .build();
    }

    // Step 정의 (Chunk 기반)
    @Bean
    public Step migrationStep(JobRepository jobRepository,
                               PlatformTransactionManager txManager,
                               ItemReader<User> userItemReader,
                               ItemProcessor<User, UserDto> userItemProcessor,
                               ItemWriter<UserDto> userItemWriter) {
        return new StepBuilder("migrationStep", jobRepository)
                .<User, UserDto>chunk(100, txManager)
                .reader(userItemReader)
                .processor(userItemProcessor)
                .writer(userItemWriter)
                .faultTolerant()
                .skipLimit(10)
                .skip(DataIntegrityViolationException.class)
                .retryLimit(3)
                .retry(TransientDataAccessException.class)
                .listener(stepExecutionListener())
                .build();
    }

    @Bean
    public JobExecutionListener jobExecutionListener() {
        return new JobExecutionListener() {
            @Override
            public void beforeJob(JobExecution jobExecution) {
                System.out.println("Job 시작: " + jobExecution.getJobInstance().getJobName());
            }

            @Override
            public void afterJob(JobExecution jobExecution) {
                System.out.printf("Job 완료: %s, 상태: %s%n",
                    jobExecution.getJobInstance().getJobName(),
                    jobExecution.getStatus());
            }
        };
    }

    @Bean
    public StepExecutionListener stepExecutionListener() {
        return new StepExecutionListener() {
            @Override
            public void beforeStep(StepExecution stepExecution) {
                System.out.println("Step 시작: " + stepExecution.getStepName());
            }

            @Override
            public ExitStatus afterStep(StepExecution stepExecution) {
                System.out.printf("Step 완료: 읽기=%d, 처리=%d, 쓰기=%d%n",
                    stepExecution.getReadCount(),
                    stepExecution.getProcessSkipCount(),
                    stepExecution.getWriteCount());
                return stepExecution.getExitStatus();
            }
        };
    }
}

@Bean
public Step cleanupStep(JobRepository jobRepository,
                         PlatformTransactionManager txManager) {
    return new StepBuilder("cleanupStep", jobRepository)
            .tasklet((contribution, chunkContext) -> {
                // 단순한 일회성 작업에 적합
                System.out.println("임시 파일 정리 중...");
                // 파일 삭제 로직
                return RepeatStatus.FINISHED;
            }, txManager)
            .build();
}

@Bean
@StepScope
public JdbcCursorItemReader<User> userCursorReader(DataSource dataSource) {
    return new JdbcCursorItemReaderBuilder<User>()
            .name("userCursorReader")
            .dataSource(dataSource)
            .sql("SELECT id, username, email, status FROM users WHERE status = 'ACTIVE' ORDER BY id")
            .rowMapper(new BeanPropertyRowMapper<>(User.class))
            .fetchSize(1000)  // 한 번에 가져올 레코드 수
            .build();
}

@Bean
@StepScope
public JdbcPagingItemReader<User> userPagingReader(DataSource dataSource) {
    Map<String, Order> sortKeys = new HashMap<>();
    sortKeys.put("id", Order.ASCENDING);

    PostgresPagingQueryProvider queryProvider = new PostgresPagingQueryProvider();
    queryProvider.setSelectClause("SELECT id, username, email, status");
    queryProvider.setFromClause("FROM users");
    queryProvider.setWhereClause("WHERE status = 'ACTIVE'");
    queryProvider.setSortKeys(sortKeys);

    return new JdbcPagingItemReaderBuilder<User>()
            .name("userPagingReader")
            .dataSource(dataSource)
            .queryProvider(queryProvider)
            .pageSize(100)
            .rowMapper(new BeanPropertyRowMapper<>(User.class))
            .build();
}

@Bean
@StepScope
public FlatFileItemReader<UserCsvDto> csvUserReader(
        @Value("#{jobParameters['inputFile']}") String inputFile) {

    return new FlatFileItemReaderBuilder<UserCsvDto>()
            .name("csvUserReader")
            .resource(new FileSystemResource(inputFile))
            .linesToSkip(1)  // 헤더 행 건너뜀
            .delimited()
            .delimiter(",")
            .names("id", "username", "email", "createdAt")
            .fieldSetMapper(new BeanWrapperFieldSetMapper<>() {{
                setTargetType(UserCsvDto.class);
            }})
            .build();
}

@Component
@StepScope
public class ApiCallItemReader implements ItemReader<UserData> {

    private final UserApiClient apiClient;
    private int page = 0;
    private List<UserData> currentPageData = new ArrayList<>();
    private int currentIndex = 0;
    private boolean exhausted = false;

    public ApiCallItemReader(UserApiClient apiClient) {
        this.apiClient = apiClient;
    }

    @Override
    public UserData read() throws Exception {
        if (exhausted) return null;

        if (currentIndex >= currentPageData.size()) {
            // 다음 페이지 로드
            currentPageData = apiClient.fetchUsers(page++, 100);
            currentIndex = 0;

            if (currentPageData.isEmpty()) {
                exhausted = true;
                return null;
            }
        }

        return currentPageData.get(currentIndex++);
    }
}

@Component
@StepScope
public class UserMigrationProcessor implements ItemProcessor<User, UserDto> {

    private static final Logger log = LoggerFactory.getLogger(UserMigrationProcessor.class);

    @Override
    public UserDto process(User user) throws Exception {
        // null 반환 = 해당 아이템 건너뜀 (skip)
        if (!isEligibleForMigration(user)) {
            log.debug("Skipping user: {}", user.getUsername());
            return null;
        }

        UserDto dto = new UserDto();
        dto.setId(user.getId());
        dto.setUsername(user.getUsername().toLowerCase().trim());
        dto.setEmail(user.getEmail().toLowerCase());
        dto.setDisplayName(formatDisplayName(user.getFirstName(), user.getLastName()));
        dto.setMigratedAt(LocalDateTime.now());

        return dto;
    }

    private boolean isEligibleForMigration(User user) {
        return user.getStatus() != null
            && "ACTIVE".equals(user.getStatus())
            && user.getEmail() != null
            && user.getEmail().contains("@");
    }

    private String formatDisplayName(String firstName, String lastName) {
        return Stream.of(firstName, lastName)
                .filter(s -> s != null && !s.isBlank())
                .collect(Collectors.joining(" "));
    }
}

@Bean
public CompositeItemProcessor<User, UserDto> compositeProcessor() {
    List<ItemProcessor<?, ?>> processors = new ArrayList<>();
    processors.add(new ValidationProcessor());
    processors.add(new EnrichmentProcessor(externalService));
    processors.add(new TransformationProcessor());

    CompositeItemProcessor<User, UserDto> composite = new CompositeItemProcessor<>();
    composite.setDelegates(processors);
    return composite;
}

@Bean
public JdbcBatchItemWriter<UserDto> userDtoWriter(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<UserDto>()
            .dataSource(dataSource)
            .sql("""
                INSERT INTO users_new (id, username, email, display_name, migrated_at)
                VALUES (:id, :username, :email, :displayName, :migratedAt)
                ON CONFLICT (id) DO UPDATE
                SET username = EXCLUDED.username,
                    email = EXCLUDED.email,
                    migrated_at = EXCLUDED.migrated_at
                """)
            .beanMapped()
            .build();
}

@Bean
public JpaItemWriter<UserDto> jpaUserWriter(EntityManagerFactory entityManagerFactory) {
    JpaItemWriter<UserDto> writer = new JpaItemWriter<>();
    writer.setEntityManagerFactory(entityManagerFactory);
    return writer;
}

@Bean
@StepScope
public FlatFileItemWriter<UserDto> csvResultWriter(
        @Value("#{jobParameters['outputFile']}") String outputFile) {

    return new FlatFileItemWriterBuilder<UserDto>()
            .name("csvResultWriter")
            .resource(new FileSystemResource(outputFile))
            .headerCallback(writer -> writer.write("id,username,email,migrated_at"))
            .delimited()
            .delimiter(",")
            .names("id", "username", "email", "migratedAt")
            .build();
}

@Bean
public CompositeItemWriter<UserDto> compositeWriter(
        JdbcBatchItemWriter<UserDto> dbWriter,
        FlatFileItemWriter<UserDto> fileWriter) {

    CompositeItemWriter<UserDto> writer = new CompositeItemWriter<>();
    writer.setDelegates(Arrays.asList(dbWriter, fileWriter));
    return writer;
}

@Configuration
public class PartitionedJobConfig {

    @Bean
    public Step masterStep(JobRepository jobRepository,
                            Partitioner partitioner,
                            Step workerStep) {
        return new StepBuilder("masterStep", jobRepository)
                .partitioner("workerStep", partitioner)
                .step(workerStep)
                .gridSize(4)  // 파티션 수 (스레드 수)
                .taskExecutor(taskExecutor())
                .build();
    }

    @Bean
    public Partitioner columnRangePartitioner(DataSource dataSource) {
        return gridSize -> {
            Map<String, ExecutionContext> result = new HashMap<>();
            int totalCount = getTotalCount(dataSource);
            int rangeSize = totalCount / gridSize;

            for (int i = 0; i < gridSize; i++) {
                ExecutionContext context = new ExecutionContext();
                context.putLong("minId", (long) i * rangeSize + 1);
                context.putLong("maxId", i == gridSize - 1 ? totalCount : (long)(i + 1) * rangeSize);
                result.put("partition" + i, context);
            }
            return result;
        };
    }

    @Bean
    @StepScope
    public JdbcPagingItemReader<User> partitionedReader(
            DataSource dataSource,
            @Value("#{stepExecutionContext['minId']}") Long minId,
            @Value("#{stepExecutionContext['maxId']}") Long maxId) {

        // 파티션 범위에 맞는 데이터만 읽기
        Map<String, Object> parameterValues = new HashMap<>();
        parameterValues.put("minId", minId);
        parameterValues.put("maxId", maxId);

        return new JdbcPagingItemReaderBuilder<User>()
                .name("partitionedReader")
                .dataSource(dataSource)
                // ... 쿼리 설정
                .parameterValues(parameterValues)
                .build();
    }

    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(8);
        executor.setQueueCapacity(25);
        executor.setThreadNamePrefix("batch-partition-");
        executor.initialize();
        return executor;
    }
}

@Bean
public Step multiThreadedStep(JobRepository jobRepository,
                               PlatformTransactionManager txManager,
                               ItemReader<User> reader,
                               ItemWriter<UserDto> writer) {
    return new StepBuilder("multiThreadedStep", jobRepository)
            .<User, UserDto>chunk(100, txManager)
            .reader(reader)     // 주의: thread-safe한 Reader 필요 (SynchronizedItemStreamReader 사용)
            .writer(writer)
            .taskExecutor(new SimpleAsyncTaskExecutor())
            .throttleLimit(4)   // 동시 실행 스레드 수
            .build();
}

// Thread-safe Reader 래핑
@Bean
public SynchronizedItemStreamReader<User> synchronizedReader(
        JdbcCursorItemReader<User> reader) {
    SynchronizedItemStreamReader<User> synchronizedReader = new SynchronizedItemStreamReader<>();
    synchronizedReader.setDelegate(reader);
    return synchronizedReader;
}

@Bean
public AsyncItemProcessor<User, UserDto> asyncProcessor(
        UserMigrationProcessor delegateProcessor) {

    AsyncItemProcessor<User, UserDto> asyncProcessor = new AsyncItemProcessor<>();
    asyncProcessor.setDelegate(delegateProcessor);
    asyncProcessor.setTaskExecutor(new SimpleAsyncTaskExecutor());
    return asyncProcessor;
}

@Bean
public AsyncItemWriter<UserDto> asyncWriter(
        JdbcBatchItemWriter<UserDto> delegateWriter) {

    AsyncItemWriter<UserDto> asyncWriter = new AsyncItemWriter<>();
    asyncWriter.setDelegate(delegateWriter);
    return asyncWriter;
}

@Bean
@StepScope
public JdbcCursorItemReader<User> dynamicReader(
        DataSource dataSource,
        @Value("#{jobParameters['startDate']}") String startDate,
        @Value("#{jobParameters['endDate']}") String endDate) {

    return new JdbcCursorItemReaderBuilder<User>()
            .name("dynamicReader")
            .dataSource(dataSource)
            .sql("SELECT * FROM users WHERE created_at BETWEEN ? AND ?")
            .preparedStatementSetter(ps -> {
                ps.setString(1, startDate);
                ps.setString(2, endDate);
            })
            .rowMapper(new BeanPropertyRowMapper<>(User.class))
            .build();
}

@Bean
public Step robustStep(JobRepository jobRepository,
                        PlatformTransactionManager txManager) {
    return new StepBuilder("robustStep", jobRepository)
            .<User, UserDto>chunk(100, txManager)
            .reader(reader())
            .processor(processor())
            .writer(writer())
            .faultTolerant()
            // Skip 설정: 최대 10건 건너뛰기 허용
            .skipLimit(10)
            .skip(ValidationException.class)
            .skip(DataIntegrityViolationException.class)
            // Retry 설정: 최대 3회 재시도
            .retryLimit(3)
            .retry(TransientDataAccessException.class)
            .retry(DeadlockLoserDataAccessException.class)
            // Skip 불가 예외
            .noSkip(FatalBatchException.class)
            .build();
}

@Component
public class UserSkipListener implements SkipListener<User, UserDto> {

    private static final Logger log = LoggerFactory.getLogger(UserSkipListener.class);

    @Override
    public void onSkipInRead(Throwable t) {
        log.error("읽기 단계에서 건너뜀: {}", t.getMessage());
    }

    @Override
    public void onSkipInProcess(User user, Throwable t) {
        log.warn("처리 건너뜀 - 사용자 ID: {}, 오류: {}", user.getId(), t.getMessage());
    }

    @Override
    public void onSkipInWrite(UserDto dto, Throwable t) {
        log.error("쓰기 건너뜀 - 사용자 ID: {}, 오류: {}", dto.getId(), t.getMessage());
    }
}

@Bean
public Job restartableJob(JobRepository jobRepository, Step step1) {
    return new JobBuilder("restartableJob", jobRepository)
            .start(step1)
            .preventRestart()  // 실패 후 재시작 방지 (기본값: 허용)
            .build();
}

// 재시작 시 같은 JobParameters 사용 → 이전 실패 지점부터 재시작
// JobLauncher에서 동일 파라미터로 실행하면 자동으로 재시작

@Configuration
@EnableScheduling
public class BatchSchedulerConfig {

    @Autowired
    private JobLauncher jobLauncher;

    @Autowired
    private Job userMigrationJob;

    @Scheduled(cron = "0 0 2 * * *")  // 매일 새벽 2시
    public void runDailyBatch() {
        try {
            JobParameters params = new JobParametersBuilder()
                    .addString("date", LocalDate.now().toString())
                    .addLong("timestamp", System.currentTimeMillis())
                    .toJobParameters();

            JobExecution execution = jobLauncher.run(userMigrationJob, params);
            System.out.println("배치 실행 상태: " + execution.getStatus());
        } catch (Exception e) {
            System.err.println("배치 실행 실패: " + e.getMessage());
        }
    }
}

@Configuration
public class QuartzBatchConfig {

    @Bean
    public JobDetail batchJobDetail() {
        return JobBuilder.newJob(BatchQuartzJob.class)
                .withIdentity("batchJob")
                .storeDurably()
                .build();
    }

    @Bean
    public Trigger batchJobTrigger() {
        return TriggerBuilder.newTrigger()
                .forJob(batchJobDetail())
                .withIdentity("batchTrigger")
                .withSchedule(CronScheduleBuilder.cronSchedule("0 0 2 * * ?"))
                .build();
    }
}

@Component
public class BatchQuartzJob implements org.quartz.Job {

    @Autowired
    private JobLauncher jobLauncher;

    @Autowired
    private Job userMigrationJob;

    @Override
    public void execute(JobExecutionContext context) throws JobExecutionException {
        try {
            JobParameters params = new JobParametersBuilder()
                    .addLong("time", System.currentTimeMillis())
                    .toJobParameters();
            jobLauncher.run(userMigrationJob, params);
        } catch (Exception e) {
            throw new JobExecutionException(e);
        }
    }
}

@RestController
@RequestMapping("/api/batch")
public class BatchController {

    @Autowired
    private JobLauncher jobLauncher;

    @Autowired
    private Job userMigrationJob;

    @PostMapping("/run")
    public ResponseEntity<String> runJob(
            @RequestParam String startDate,
            @RequestParam String endDate) {
        try {
            JobParameters params = new JobParametersBuilder()
                    .addString("startDate", startDate)
                    .addString("endDate", endDate)
                    .addLong("timestamp", System.currentTimeMillis())
                    .toJobParameters();

            JobExecution execution = jobLauncher.run(userMigrationJob, params);
            return ResponseEntity.ok("Job 실행 ID: " + execution.getId()
                + ", 상태: " + execution.getStatus());
        } catch (Exception e) {
            return ResponseEntity.internalServerError()
                .body("Job 실행 실패: " + e.getMessage());
        }
    }

    @GetMapping("/status/{executionId}")
    public ResponseEntity<String> getJobStatus(@PathVariable Long executionId) {
        // JobExplorer로 실행 이력 조회
        return ResponseEntity.ok("구현 필요");
    }
}

@SpringBatchTest
@SpringBootTest(classes = {UserMigrationJobConfig.class, TestBatchConfig.class})
@ActiveProfiles("test")
class UserMigrationJobTest {

    @Autowired
    private JobLauncherTestUtils jobLauncherTestUtils;

    @Autowired
    private JobRepositoryTestUtils jobRepositoryTestUtils;

    @BeforeEach
    void clearMetadata() {
        jobRepositoryTestUtils.removeJobExecutions();
    }

    @Test
    void testCompleteJob() throws Exception {
        JobExecution jobExecution = jobLauncherTestUtils.launchJob(
            new JobParametersBuilder()
                .addString("date", "2026-03-17")
                .toJobParameters()
        );

        assertThat(jobExecution.getStatus()).isEqualTo(BatchStatus.COMPLETED);
        assertThat(jobExecution.getExitStatus()).isEqualTo(ExitStatus.COMPLETED);
    }

    @Test
    void testStep() throws Exception {
        JobExecution jobExecution = jobLauncherTestUtils.launchStep("migrationStep");

        StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next();
        assertThat(stepExecution.getStatus()).isEqualTo(BatchStatus.COMPLETED);
        assertThat(stepExecution.getWriteCount()).isGreaterThan(0);
    }
}

@RunWith(SpringRunner.class)
@SpringBootTest
@TestExecutionListeners({
    DependencyInjectionTestExecutionListener.class,
    StepScopeTestExecutionListener.class
})
class ItemReaderTest {

    @Autowired
    private JdbcCursorItemReader<User> userReader;

    public StepExecution getStepExecution() {
        StepExecution execution = MetaDataInstanceFactory.createStepExecution();
        execution.getExecutionContext().putString("inputFile", "classpath:test-users.csv");
        return execution;
    }

    @Test
    void testReader() throws Exception {
        List<User> users = new ArrayList<>();
        User user;
        while ((user = userReader.read()) != null) {
            users.add(user);
        }
        assertThat(users).isNotEmpty();
    }
}

@SpringBatchTest
@SpringBootTest
@Testcontainers
class BatchIntegrationTest {

    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:15");

    @DynamicPropertySource
    static void setProps(DynamicPropertyRegistry registry) {
        registry.add("spring.datasource.url", postgres::getJdbcUrl);
        registry.add("spring.datasource.username", postgres::getUsername);
        registry.add("spring.datasource.password", postgres::getPassword);
    }

    @Autowired
    private JobLauncherTestUtils jobLauncherTestUtils;

    @Autowired
    private UserRepository userRepository;

    @Test
    void testFullMigrationPipeline() throws Exception {
        // Given: 테스트 데이터 삽입
        insertTestUsers(100);

        // When: 배치 실행
        JobExecution execution = jobLauncherTestUtils.launchJob();

        // Then: 결과 검증
        assertThat(execution.getStatus()).isEqualTo(BatchStatus.COMPLETED);
        assertThat(userRepository.countByMigratedTrue()).isEqualTo(100);
    }
}

management:
  endpoints:
    web:
      exposure:
        include: health,info,metrics,batch
  endpoint:
    batch:
      enabled: true

@Configuration
public class BatchMetricsConfig {

    @Bean
    public BatchMetrics batchMetrics(MeterRegistry meterRegistry,
                                      JobRepository jobRepository) {
        return new BatchMetrics(meterRegistry);
    }
}

# Grafana 대시보드용 주요 메트릭
# spring.batch.job.* - Job 실행 시간, 성공/실패 수
# spring.batch.step.* - Step별 읽기/처리/쓰기 건수