SpringBatchの紹介
1. 概要
このチュートリアルでは、Springバッチの実用的なコードに焦点を当てたイントロを見ていきます。 Spring Batchは、ジョブを堅牢に実行するために設計された処理フレームワークです。
現在のバージョン4.3は、Spring5およびJava8をサポートしています。 また、バッチ処理用の新しいJava仕様であるJSR-352にも対応しています。
ここにフレームワークのいくつかの興味深く実用的なユースケースがあります。
2. ワークフローの基本
Spring Batchは、ジョブリポジトリがジョブのスケジューリングと対話の作業を行う従来のバッチアーキテクチャに従います。
ジョブには複数のステップを含めることができます。 そして、すべてのステップは通常、データの読み取り、処理、書き込みのシーケンスに従います。
そしてもちろん、フレームワークは、ジョブリポジトリに sqlite を使用して、ここでの面倒な作業のほとんどを実行します。特に、ジョブを処理する低レベルの永続化作業に関してはそうです。
2.1. ユースケースの例
ここで取り組む簡単なユースケースは、一部の金融取引データをCSVからXMLに移行することです。
入力ファイルの構造は非常に単純です。
これには、ユーザー名、ユーザーID、トランザクションの日付、および金額で構成される1行あたりのトランザクションが含まれます。
username, userid, transaction_date, transaction_amount
devendra, 1234, 31/10/2015, 10000
john, 2134, 3/12/2015, 12321
robin, 2134, 2/02/2015, 23411
3. Maven POM
このプロジェクトに必要な依存関係は、Spring Core、Spring Batch、および sqlitejdbcコネクタです。
<!-- SQLite database driver -->
<dependency>
<groupId>org.xerial</groupId>
<artifactId>sqlite-jdbc</artifactId>
<version>3.15.1</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-oxm</artifactId>
<version>5.3.0</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jdbc</artifactId>
<version>5.3.0</version>
</dependency>
<dependency>
<groupId>org.springframework.batch</groupId>
<artifactId>spring-batch-core</artifactId>
<version>4.3.0</version>
</dependency>
4. Spring BatchConfig
最初に行うことは、XMLを使用してSpringバッチを構成することです。
<!-- connect to SQLite database -->
<bean id="dataSource"
class="org.springframework.jdbc.datasource.DriverManagerDataSource">
<property name="driverClassName" value="org.sqlite.JDBC" />
<property name="url" value="jdbc:sqlite:repository.sqlite" />
<property name="username" value="" />
<property name="password" value="" />
</bean>
<!-- create job-meta tables automatically -->
<jdbc:initialize-database data-source="dataSource">
<jdbc:script
location="org/springframework/batch/core/schema-drop-sqlite.sql" />
<jdbc:script location="org/springframework/batch/core/schema-sqlite.sql" />
</jdbc:initialize-database>
<!-- stored job-meta in memory -->
<!--
<bean id="jobRepository"
class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean">
<property name="transactionManager" ref="transactionManager" />
</bean>
-->
<!-- stored job-meta in database -->
<bean id="jobRepository"
class="org.springframework.batch.core.repository.support.JobRepositoryFactoryBean">
<property name="dataSource" ref="dataSource" />
<property name="transactionManager" ref="transactionManager" />
<property name="databaseType" value="sqlite" />
</bean>
<bean id="transactionManager" class=
"org.springframework.batch.support.transaction.ResourcelessTransactionManager" />
<bean id="jobLauncher"
class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
<property name="jobRepository" ref="jobRepository" />
</bean>
もちろん、Java構成も利用できます。
@Configuration
@EnableBatchProcessing
public class SpringConfig {
@Value("org/springframework/batch/core/schema-drop-sqlite.sql")
private Resource dropReopsitoryTables;
@Value("org/springframework/batch/core/schema-sqlite.sql")
private Resource dataReopsitorySchema;
@Bean
public DataSource dataSource() {
DriverManagerDataSource dataSource = new DriverManagerDataSource();
dataSource.setDriverClassName("org.sqlite.JDBC");
dataSource.setUrl("jdbc:sqlite:repository.sqlite");
return dataSource;
}
@Bean
public DataSourceInitializer dataSourceInitializer(DataSource dataSource)
throws MalformedURLException {
ResourceDatabasePopulator databasePopulator =
new ResourceDatabasePopulator();
databasePopulator.addScript(dropReopsitoryTables);
databasePopulator.addScript(dataReopsitorySchema);
databasePopulator.setIgnoreFailedDrops(true);
DataSourceInitializer initializer = new DataSourceInitializer();
initializer.setDataSource(dataSource);
initializer.setDatabasePopulator(databasePopulator);
return initializer;
}
private JobRepository getJobRepository() throws Exception {
JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
factory.setDataSource(dataSource());
factory.setTransactionManager(getTransactionManager());
factory.afterPropertiesSet();
return (JobRepository) factory.getObject();
}
private PlatformTransactionManager getTransactionManager() {
return new ResourcelessTransactionManager();
}
public JobLauncher getJobLauncher() throws Exception {
SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
jobLauncher.setJobRepository(getJobRepository());
jobLauncher.afterPropertiesSet();
return jobLauncher;
}
}
5. SpringBatchジョブ構成
次に、CSVからXMLへの作業の職務記述書を書いてみましょう。
<import resource="spring.xml" />
<bean id="record" class="com.baeldung.spring_batch_intro.model.Transaction"></bean>
<bean id="itemReader"
class="org.springframework.batch.item.file.FlatFileItemReader">
<property name="resource" value="input/record.csv" />
<property name="lineMapper">
<bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
<property name="lineTokenizer">
<bean class=
"org.springframework.batch.item.file.transform.DelimitedLineTokenizer">
<property name="names" value="username,userid,transactiondate,amount" />
</bean>
</property>
<property name="fieldSetMapper">
<bean class="com.baeldung.spring_batch_intro.service.RecordFieldSetMapper" />
</property>
</bean>
</property>
</bean>
<bean id="itemProcessor"
class="com.baeldung.spring_batch_intro.service.CustomItemProcessor" />
<bean id="itemWriter"
class="org.springframework.batch.item.xml.StaxEventItemWriter">
<property name="resource" value="file:xml/output.xml" />
<property name="marshaller" ref="recordMarshaller" />
<property name="rootTagName" value="transactionRecord" />
</bean>
<bean id="recordMarshaller" class="org.springframework.oxm.jaxb.Jaxb2Marshaller">
<property name="classesToBeBound">
<list>
<value>com.baeldung.spring_batch_intro.model.Transaction</value>
</list>
</property>
</bean>
<batch:job id="firstBatchJob">
<batch:step id="step1">
<batch:tasklet>
<batch:chunk reader="itemReader" writer="itemWriter"
processor="itemProcessor" commit-interval="10">
</batch:chunk>
</batch:tasklet>
</batch:step>
</batch:job>
同様のJavaベースのジョブ構成は次のとおりです。
public class SpringBatchConfig {
@Autowired
private JobBuilderFactory jobs;
@Autowired
private StepBuilderFactory steps;
@Value("input/record.csv")
private Resource inputCsv;
@Value("file:xml/output.xml")
private Resource outputXml;
@Bean
public ItemReader<Transaction> itemReader()
throws UnexpectedInputException, ParseException {
FlatFileItemReader<Transaction> reader = new FlatFileItemReader<Transaction>();
DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
String[] tokens = { "username", "userid", "transactiondate", "amount" };
tokenizer.setNames(tokens);
reader.setResource(inputCsv);
DefaultLineMapper<Transaction> lineMapper =
new DefaultLineMapper<Transaction>();
lineMapper.setLineTokenizer(tokenizer);
lineMapper.setFieldSetMapper(new RecordFieldSetMapper());
reader.setLineMapper(lineMapper);
return reader;
}
@Bean
public ItemProcessor<Transaction, Transaction> itemProcessor() {
return new CustomItemProcessor();
}
@Bean
public ItemWriter<Transaction> itemWriter(Marshaller marshaller)
throws MalformedURLException {
StaxEventItemWriter<Transaction> itemWriter =
new StaxEventItemWriter<Transaction>();
itemWriter.setMarshaller(marshaller);
itemWriter.setRootTagName("transactionRecord");
itemWriter.setResource(outputXml);
return itemWriter;
}
@Bean
public Marshaller marshaller() {
Jaxb2Marshaller marshaller = new Jaxb2Marshaller();
marshaller.setClassesToBeBound(new Class[] { Transaction.class });
return marshaller;
}
@Bean
protected Step step1(ItemReader<Transaction> reader,
ItemProcessor<Transaction, Transaction> processor,
ItemWriter<Transaction> writer) {
return steps.get("step1").<Transaction, Transaction> chunk(10)
.reader(reader).processor(processor).writer(writer).build();
}
@Bean(name = "firstBatchJob")
public Job job(@Qualifier("step1") Step step1) {
return jobs.get("firstBatchJob").start(step1).build();
}
}
構成全体ができたので、それを分解して説明を始めましょう。
5.1. ItemReaderを使用してデータを読み取りオブジェクトを作成する
まず、 record.csv からデータを読み取り、それをTransactionオブジェクトに変換するcvsFileItemReaderを構成しました。
@SuppressWarnings("restriction")
@XmlRootElement(name = "transactionRecord")
public class Transaction {
private String username;
private int userId;
private LocalDateTime transactionDate;
private double amount;
/* getters and setters for the attributes */
@Override
public String toString() {
return "Transaction [username=" + username + ", userId=" + userId
+ ", transactionDate=" + transactionDate + ", amount=" + amount
+ "]";
}
}
そのために、カスタムマッパーを使用します。
public class RecordFieldSetMapper implements FieldSetMapper<Transaction> {
public Transaction mapFieldSet(FieldSet fieldSet) throws BindException {
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("d/M/yyy");
Transaction transaction = new Transaction();
transaction.setUsername(fieldSet.readString("username"));
transaction.setUserId(fieldSet.readInt(1));
transaction.setAmount(fieldSet.readDouble(3));
String dateString = fieldSet.readString(2);
transaction.setTransactionDate(LocalDate.parse(dateString, formatter).atStartOfDay());
return transaction;
}
}
5.2. ItemProcessorを使用したデータの処理
独自のアイテムプロセッサCustomItemProcessorを作成しました。 これは、トランザクションオブジェクトに関連するものは何も処理しません。
リーダーからライターに送られる元のオブジェクトを渡すだけです。
public class CustomItemProcessor implements ItemProcessor<Transaction, Transaction> {
public Transaction process(Transaction item) {
return item;
}
}
5.3. ItemWriterを使用したFSへのオブジェクトの書き込み
最後に、このtransactionをxml/output.xmlにあるXMLファイルに保存します。
<bean id="itemWriter"
class="org.springframework.batch.item.xml.StaxEventItemWriter">
<property name="resource" value="file:xml/output.xml" />
<property name="marshaller" ref="recordMarshaller" />
<property name="rootTagName" value="transactionRecord" />
</bean>
5.4. バッチジョブの構成
したがって、 batch:job 構文を使用して、ドットをジョブに接続するだけです。
commit-intervalに注意してください。 これは、バッチをitemWriterにコミットする前にメモリに保持されるトランザクションの数です。
その時点まで(または入力データの終わりに遭遇するまで)、トランザクションをメモリに保持します。
<batch:job id="firstBatchJob">
<batch:step id="step1">
<batch:tasklet>
<batch:chunk reader="itemReader" writer="itemWriter"
processor="itemProcessor" commit-interval="10">
</batch:chunk>
</batch:tasklet>
</batch:step>
</batch:job>
5.5. バッチジョブの実行
それでは、すべてをセットアップして実行しましょう。
public class App {
public static void main(String[] args) {
// Spring Java config
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
context.register(SpringConfig.class);
context.register(SpringBatchConfig.class);
context.refresh();
JobLauncher jobLauncher = (JobLauncher) context.getBean("jobLauncher");
Job job = (Job) context.getBean("firstBatchJob");
System.out.println("Starting the batch job");
try {
JobExecution execution = jobLauncher.run(job, new JobParameters());
System.out.println("Job Status : " + execution.getStatus());
System.out.println("Job completed");
} catch (Exception e) {
e.printStackTrace();
System.out.println("Job failed");
}
}
}
6. 結論
この記事では、 Spring Batch の操作方法と、簡単なユースケースでの使用方法について説明しました。
バッチ処理パイプラインを簡単に開発する方法と、読み取り、処理、書き込みのさまざまな段階をカスタマイズする方法を示しています。
この記事の完全な実装は、GitHubプロジェクトにあります。 これはEclipseベースのプロジェクトであるため、そのままインポートして実行するのは簡単です。