はじめに
本記事はゼロからSpring Batchを説明し、活用して環境構築を行ってバッチ処理を実装してみます。
処理流れの全体としては、バッチプロセス
が動いたら、あるjSONファイルからデータを読み込んでMySQLのデータベースに書き出す。
SpringBatchとは
Spring BatchはSpringの子どものプロジェクトである。Spring BatchはSpringを基に、バッチアプリケーションフレームワークの一つである。
SpringBatchの構成
Spring Batchの主な構成要素と処理全体の流れについて説明をする。 また、ジョブの実行状況などのメタデータがどのように管理されているかについても説明する。
Spring Batchの主な構成要素と処理全体の流れ(チャンクモデル)を下図に示す。
| 構成要素 | 役割 |
|---|---|
| Job | Spring Batchにおけるバッチアプリケーションの一連の処理をまとめた1実行単位。 |
| Step | Jobを構成する処理の単位。1つのJobに1～N個のStepをもたせることが可能。 |
| JobLauncher | Jobを起動するためのインターフェース。 |
| ItemReader | チャンクモデルを実装する際に、データの入力／加工／出力の3つに分割するためのインターフェース。 |
| ItemProcessor | チャンクモデルを実装する際に、データの入力／加工／出力の3つに分割するためのインターフェース。 |
| ItemWriter | チャンクモデルを実装する際に、データの入力／加工／出力の3つに分割するためのインターフェース。 |
| JobRepository | JobやStepの状況を管理する機構。これらの管理情報は、Spring Batchが規定するテーブルスキーマを元にデータベース上に永続化される。 |
環境構築
- SpringBatch
- SpringBoot
- Java
- Grade
本記事では、Spring Initializrを活用してプロジェクトを作りました。以下の通りにやりましょう。
ソース実装
こらから、プロジェクトを作って説明します。
ライブラリ導入
このファイルがライブラリ依頼となります。
Build.gralde
buildscript {
ext {
springBootVersion = '2.0.4.RELEASE'
}
repositories {
mavenCentral()
}
dependencies {
classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
}
}
apply plugin: 'java'
apply plugin: 'idea'
apply plugin: 'org.springframework.boot'
apply plugin: 'io.spring.dependency-management'
group = 'com.demo'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = 1.8
repositories {
mavenCentral()
}
dependencies {
compile('org.springframework.boot:spring-boot-starter-batch')
compile('org.springframework.boot:spring-boot-starter-jdbc')
compile("org.springframework.boot:spring-boot-starter-data-jpa")
compile group: 'com.fasterxml.jackson.datatype', name: 'jackson-datatype-joda', version: '2.9.4'
compile group: 'org.jadira.usertype', name: 'usertype.core', version: '6.0.1.GA'
compile group: 'mysql', name: 'mysql-connector-java', version: '6.0.6',
testCompile('org.springframework.boot:spring-boot-starter-test')
testCompile('org.springframework.batch:spring-batch-test')
}
SpringBatchの配置
まずは、グローバル設定を行います。後はJOBの配置をします。
SpringBatchConfiguration.java
@Configuration
@EnableAutoConfiguration
@EnableBatchProcessing(modular = true)
public class SpringBatchConfiguration {
@Bean
public ApplicationContextFactory firstJobContext() {
return new GenericApplicationContextFactory(FirstJobConfiguration.class);
}
@Bean
public ApplicationContextFactory secondJobContext() {
return new GenericApplicationContextFactory(SecondJobConfiguration.class);
}
}
エンティティクラスを作成
Json Dataのフォーマットの通りです。
Message.java
@Entity
@Table(name = "message")
public class Message {
@Id
@Column(name = "object_id", nullable = false)
private String objectId;
@Column(name = "content")
private String content;
@Column(name = "last_modified_time")
private LocalDateTime lastModifiedTime;
@Column(name = "created_time")
private LocalDateTime createdTime;
}
Jobを作成
MessageMigrationJobConfiguration.java
public class MessageMigrationJobConfiguration {
@Autowired
private JobBuilderFactory jobBuilderFactory;
@Autowired
private StepBuilderFactory stepBuilderFactory
@Autowired
private EntityManagerFactory entityManager;
// Jobを作成
@Bean
public Job messageMigrationJob(@Qualifier("messageMigrationStep") Step messageMigrationStep) {
return jobBuilderFactory.get("messageMigrationJob")
.start(messageMigrationStep)
.build();
}
// Stepを作成
@Bean
public Step messageMigrationStep(@Qualifier("jsonMessageReader") FlatFileItemReader<Message> jsonMessageReader,
@Qualifier("messageItemWriter") JpaItemWriter<Message> messageItemWriter,
@Qualifier("errorWriter") Writer errorWriter) {
return stepBuilderFactory.get("messageMigrationStep")
.<Message, Message>chunk(CHUNK_SIZE)
.reader(jsonMessageReader).faultTolerant().skip(JsonParseException.class).skipLimit(SKIP_LIMIT)
.listener(new MessageItemReadListener(errorWriter))
.writer(messageItemWriter).faultTolerant().skip(Exception.class).skipLimit(SKIP_LIMIT)
.listener(new MessageWriteListener())
.build();
}
// Readerを作成
@Bean
public FlatFileItemReader<Message> jsonMessageReader() {
FlatFileItemReader<Message> reader = new FlatFileItemReader<>();
// 実際のJson Dataファイルパスを入れる
reader.setResource(new FileSystemResource(new File(MESSAGE_FILE)));
reader.setLineMapper(new MessageLineMapper());
return reader;
}
// Writerを作成
@Bean
public JpaItemWriter<Message> messageItemWriter() {
JpaItemWriter<Message> writer = new JpaItemWriter<>();
writer.setEntityManagerFactory(entityManager);
return writer;
}
}
データプロセスメソッドを実装
ここでLineMapperを使ってtxtソースファイルから1行ずつデータを読み込んでMessageオブジェクトを転換します。
MessageLineMapper.java
public class MessageLineMapper implements LineMapper<Message> {
private MappingJsonFactory factory = new MappingJsonFactory();
@Override
public Message mapLine(String line, int lineNumber) throws Exception {
JsonParser parser = factory.createParser(line);
Map<String, Object> map = (Map) parser.readValueAs(Map.class);
Message message = new Message();
... // TODO データ転換ロジック
return message;
}
}
```
#### propertiesファイルを作成
```java:application.properties
spring.datasource.url=jdbc:mysql://database
spring.datasource.username=username
spring.datasource.password=password
spring.datasource.driverClassName=com.mysql.cj.jdbc.Driver
spring.jpa.database-platform=org.hibernate.dialect.MySQLDialect
spring.jpa.show-sql=true
spring.jpa.properties.jadira.usertype.autoRegisterUserTypes=true
spring.jackson.serialization.write-dates-as-timestamps=false
spring.batch.initialize-schema=ALWAYS
spring.jpa.hibernate.ddl-auto=update
```
#### Listenerを実装
エラーが発生する時に、リスナーがエラーメッセージをファイルに出力する。
1.ReadListener
```java:MessageItemReadListener.java
public class MessageItemReadListener implements ItemReadListener<Message> {
private Writer errorWriter;
public MessageItemReadListener(Writer errorWriter) {
this.errorWriter = errorWriter;
}
@Override
public void beforeRead() {
}
@Override
public void afterRead(Message item) {
}
@Override
public void onReadError(Exception ex) {
errorWriter.write(format("%s%n", ex.getMessage()));
}
}
```
2.WriteListener
```java:MessageWriteListener.java
public class MessageWriteListener implements ItemWriteListener<Message> {
@Autowired
private Writer errorWriter;
@Override
public void beforeWrite(List<? extends Message> items) {
}
@Override
public void afterWrite(List<? extends Message> items) {
}
@Override
public void onWriteError(Exception exception, List<? extends Message> items) {
errorWriter.write(format("%s%n", exception.getMessage()));
for (Message message : items) {
errorWriter.write(format("Failed writing message id: %s", message.getObjectId()));
}
}
}
```
#### Jobを実行
まずはテストメソッドを作ってます。
```java:SpringBatchTest.java
public static void main(String[] args) {
String jobName = args[0];
try {
ConfigurableApplicationContext context = SpringApplication.run(ZuociBatchApplication.class, args);
JobRegistry jobRegistry = context.getBean(JobRegistry.class);
Job job = jobRegistry.getJob(jobName);
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
JobExecution jobExecution = jobLauncher.run(job, createJobParams());
if (!jobExecution.getExitStatus().equals(ExitStatus.COMPLETED)) {
throw new RuntimeException(format("%s Job execution failed.", jobName));
}
} catch (Exception e) {
throw new RuntimeException(format("%s Job execution failed.", jobName));
}
}
private static JobParameters createJobParams() {
return new JobParametersBuilder().addDate("date", new Date()).toJobParameters();
}
```
プロジェクトをjarファイルにして以下のcommandで実行してみよう。
> java -jar YOUR_BATCH_NAME.jar YOUR_JOB_NAME
#### 実行結果
実行してうまくいけば、以下のようなテーブルとデータを作成するべきだ。
また、詳しい情報（job名、状態、エラーメッセージなど）は以下の通りです。
### 最後に
最後まで読んでいただき、ありがとうございます。
おかしいと思う部分は遠慮なくご指摘いただければと思います。
よろしくお願いします。
参考資料：
１．https://terasoluna-batch.github.io/guideline/5.0.0.RELEASE/ja/Ch02_SpringBatchArchitecture.html
２．https://spring.io/projects/spring-batch