spring-batch

Spring Batch

完全なリファレンス：スキップ／再試行の設定、パーティショニング、リスナー、JobLauncherTestUtils を使ったテスト、複合ライター、非同期処理については、advanced.md を参照してください。

詳細なナレッジ：包括的なドキュメントについては、technology: spring-boot、topic: batch を指定して mcp__documentation__fetch_docs を使用してください。

クイックスタート

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    @Bean
    public Job importJob(JobRepository jobRepository, Step step1) {
        return new JobBuilder("importJob", jobRepository)
            .incrementer(new RunIdIncrementer())
            .start(step1)
            .build();
    }

    @Bean
    public Step step1(JobRepository jobRepository,
                      PlatformTransactionManager transactionManager,
                      ItemReader<InputData> reader,
                      ItemProcessor<InputData, OutputData> processor,
                      ItemWriter<OutputData> writer) {
        return new StepBuilder("step1", jobRepository)
            .<InputData, OutputData>chunk(100, transactionManager)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .build();
    }
}

---

コアコンポーネント

Job

@Bean
public Job complexJob(JobRepository jobRepository,
                      Step extractStep, Step transformStep, Step loadStep) {
    return new JobBuilder("etlJob", jobRepository)
        .incrementer(new RunIdIncrementer())
        .validator(jobParametersValidator())
        .listener(jobExecutionListener())
        .start(extractStep)
        .next(transformStep)
        .next(loadStep)
        .build();
}

// Job with decision
@Bean
public Job conditionalJob(JobRepository jobRepository,
                          Step step1, Step step2, Step errorStep,
                          JobExecutionDecider decider) {
    return new JobBuilder("conditionalJob", jobRepository)
        .start(step1)
        .next(decider)
        .on("COMPLETED").to(step2)
        .from(decider).on("FAILED").to(errorStep)
        .end()
        .build();
}

Step - チャンク処理

@Bean
public Step chunkStep(JobRepository jobRepository,
                      PlatformTransactionManager txManager) {
    return new StepBuilder("chunkStep", jobRepository)
        .<Person, Person>chunk(100, txManager)
        .reader(reader())
        .processor(processor())
        .writer(writer())
        .faultTolerant()
        .skipLimit(10)
        .skip(ValidationException.class)
        .retryLimit(3)
        .retry(TransientException.class)
        .build();
}

// Tasklet Step (for simple operations)
@Bean
public Step taskletStep(JobRepository jobRepository,
                        PlatformTransactionManager txManager) {
    return new StepBuilder("taskletStep", jobRepository)
        .tasklet((contribution, chunkContext) -> {
            cleanupService.cleanup();
            return RepeatStatus.FINISHED;
        }, txManager)
        .build();
}

---

ItemReader

FlatFileItemReader

@Bean
public FlatFileItemReader<Person> csvReader() {
    return new FlatFileItemReaderBuilder<Person>()
        .name("personReader")
        .resource(new ClassPathResource("data/input.csv"))
        .delimited()
        .delimiter(",")
        .names("firstName", "lastName", "email", "age")
        .linesToSkip(1)  // Skip header
        .fieldSetMapper(new BeanWrapperFieldSetMapper<>() {{
            setTargetType(Person.class);
        }})
        .build();
}

JdbcPagingItemReader

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

    return new JdbcPagingItemReaderBuilder<Person>()
        .name("pagingReader")
        .dataSource(dataSource)
        .selectClause("SELECT id, first_name, last_name, email")
        .fromClause("FROM persons")
        .whereClause("WHERE status = :status")
        .parameterValues(Map.of("status", "ACTIVE"))
        .sortKeys(sortKeys)
        .rowMapper(new BeanPropertyRowMapper<>(Person.class))
        .pageSize(100)
        .build();
}

JpaPagingItemReader

@Bean
public JpaPagingItemReader<Person> jpaReader(EntityManagerFactory emf) {
    return new JpaPagingItemReaderBuilder<Person>()
        .name("jpaReader")
        .entityManagerFactory(emf)
        .queryString("SELECT p FROM Person p WHERE p.status = :status")
        .parameterValues(Map.of("status", Status.ACTIVE))
        .pageSize(100)
        .build();
}

---

ItemProcessor

@Component
public class PersonProcessor implements ItemProcessor<Person, Person> {

    @Override
    public Person process(Person person) throws Exception {
        // Return null to filter out item
        if (!isValid(person)) {
            return null;
        }

        // Transform
        person.setEmail(person.getEmail().toLowerCase());
        person.setFullName(person.getFirstName() + " " + person.getLastName());
        return person;
    }
}

// Composite processor
@Bean
public CompositeItemProcessor<Person, Person> compositeProcessor() {
    return new CompositeItemProcessorBuilder<Person, Person>()
        .delegates(List.of(
            validationProcessor(),
            transformationProcessor(),
            enrichmentProcessor()
        ))
        .build();
}

---

ItemWriter

JdbcBatchItemWriter

@Bean
public JdbcBatchItemWriter<Person> jdbcWriter(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<Person>()
        .dataSource(dataSource)
        .sql("INSERT INTO persons (first_name, last_name, email) VALUES (:firstName, :lastName, :email)")
        .beanMapped()
        .build();
}

JpaItemWriter

@Bean
public JpaItemWriter<Person> jpaWriter(EntityManagerFactory emf) {
    JpaItemWriter<Person> writer = new JpaItemWriter<>();
    writer.setEntityManagerFactory(emf);
    writer.setUsePersist(true);  // false = merge
    return writer;
}

FlatFileItemWriter

@Bean
public FlatFileItemWriter<Person> csvWriter() {
    return new FlatFileItemWriterBuilder<Person>()
        .name("personWriter")
        .resource(new FileSystemResource("output/persons.csv"))
        .delimited()
        .delimiter(",")
        .names("firstName", "lastName", "email")
        .headerCallback(writer -> writer.write("First Name,Last Name,Email"))
        .build();
}

---

ジョブパラメーター

@Bean
@StepScope
public FlatFileItemReader<Person> parameterizedReader(
        @Value("#{jobParameters['inputFile']}") String inputFile) {
    return new FlatFileItemReaderBuilder<Person>()
        .name("reader")
        .resource(new FileSystemResource(inputFile))
        .delimited()
        .names("firstName", "lastName", "email")
        .targetType(Person.class)
        .build();
}

// Running job with parameters
@Service
@RequiredArgsConstructor
public class JobLauncherService {

    private final JobLauncher jobLauncher;
    private final Job importJob;

    public void runJob(String inputFile, LocalDate date) throws Exception {
        JobParameters params = new JobParametersBuilder()
            .addString("inputFile", inputFile)
            .addLocalDate("date", date)
            .addLong("timestamp", System.currentTimeMillis())
            .toJobParameters();

        JobExecution execution = jobLauncher.run(importJob, params);
        log.info("Job status: {}", execution.getStatus());
    }
}

---

このスキルを使用すべきでないケース

• リアルタイム処理 - ストリーミング（Kafka Streams、Flink）を使用してください
• 単純なスケジュールタスク - 代わりに spring-scheduling を使用してください
• マイクロサービスのデータ同期 - メッセージングを使ったイベント駆動方式を検討してください
• 小規模なデータセット - バッチ処理のオーバーヘッドが正当化できない場合があります

アンチパターン

アンチパターン	問題	解決策
チャンクサイズが大きすぎる	メモリの問題、トランザクションの長期化	チャンクサイズを調整する（100～1000）
スキップポリシーがない	1つのエラーでジョブが停止する	想定されるエラーに対してスキップを設定する
状態を持つ ItemProcessor	スレッドセーフではない問題	プロセッサーをステートレスにする
ジョブパラメーターを無視する	失敗したジョブを再起動できない	識別用パラメーターを含める
モニタリングがない	気付かれないまま失敗する	JobExecutionListener を設定する
大量データに対して単一スレッドを使う	処理が遅い	パーティショニングを使用する

よくある問題のトラブルシューティング

問題	診断	修正方法
ジョブが開始しない	ジョブリポジトリを確認する	データベーススキーマを確認する
再起動時にジョブが失敗する	ジョブパラメーターを確認する	RunIdIncrementer を追加する
チャンク処理が遅い	コミット間隔を確認する	チャンクサイズを調整する
メモリの問題	ヒープ使用量を監視する	チャンクサイズを小さくし、データをストリーミングする
スキップが機能しない	スキップポリシーを確認する	スキップ可能な例外を設定する

ベストプラクティス

• ✅ 大量データにはチャンク処理を使用する
• ✅ 障害耐性のためにスキップ／再試行を設定する
• ✅ 並列処理にはパーティショニングを使用する
• ✅ モニタリング用のリスナーを実装する
• ✅ JobLauncherTestUtils を使ってテストする
• ❌ チャンクサイズを大きくしすぎない
• ❌ エラーを黙って無視しない
• ❌ 再起動用のジョブパラメーターを忘れない

リファレンスドキュメント

• Spring Batch