IK.AM

@making's tech note

【Spring Advent Calendar 2013 1日目 Javaバッチフレームワークの決定版。jBatch(JSR-352) on Spring Batch 3 spadc13

🗃 {Programming/Java/Spring/AdventCalendar/2013}
🗓 Updated at 2013-12-01T02:40:39Z  🗓 Created at 2013-12-01T02:40:39Z   🌎 English Page

今年もSpring Advent Calendarが始まりました。一日目の今日はJSR-352 on Spring Batch 3について話します。ハッシュタグは#spadc13でお願いします。

Java EE7でjBatchというバッチ処理(JSR-352)がサポートされました。 内容はというとSpring Batchの劣○版ですね。はい。 APIは決まりましたが、実際にバッチ処理を書くためのサポートがほとんどないです。

ただ、強いてメリットを上げるならAPIがシンプルで学習コストが低そうに見えることでしょうか。

処理方式は大きく分けて2つ

  • 単純に処理を書き下していくBatchletパターン
  • ETL処理をそれぞれ分けたItemReader/ItemProcessor/ItemWriterパターン

前者はコマンドパターン、後者をETLパターンとでも言いますか。

JSR-352の元になったのはSpring Batchですが、Spring Batchはバージョン3からJSR-352に(半分ほど)対応します。

Spring Batch 3を使うとJava標準のAPIでバッチ処理を書けるようになりますし、バッチ処理をAPサーバー上じゃなくてもスタンドアローンで実行できるようになります（もちろんAPサーバー上でもOK）。さらにはSpring Batchで用意されている様々なコンポーネント群（File連携、DB連携、MQ連携、Validationなど）を利用できるようになります。

これはJavaのバッチフレームワークの決定版ではないかと思います。

今回はSpring Batch3を使ったJSR-352のサンプルをいくつかあげてみます。

プロジェクト設定

  • pom.xml

    現時点では3はまだ正式リリース前なのでSpring Batchは3.0.0.M2を使用します。またデフォルト設定ではcommons-dbcpとhsqldbが必要です。

      <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
      <modelVersion>4.0.0</modelVersion>
      <groupId>com.example.jsr352</groupId>
      <artifactId>hello-jsr352-spring</artifactId>
      <version>1.0.0-SNAPSHOT</version>
  
      <repositories>
          <repository>
              <snapshots>
                  <enabled>false</enabled>
              </snapshots>
              <id>spring-milestones</id>
              <name>Spring Milestones</name>
              <url>http://repo.spring.io/milestone</url>
          </repository>
      </repositories>
  
      <build>
          <plugins>
              <plugin>
                  <artifactId>maven-compiler-plugin</artifactId>
                  <version>2.5.1</version>
                  <configuration>
                      <source>1.7</source>
                      <target>1.7</target>
                  </configuration>
              </plugin>
          </plugins>
      </build>
  
      <dependencies>
          <dependency>
              <groupId>org.springframework</groupId>
              <artifactId>spring-context</artifactId>
              <version>3.2.5.RELEASE</version>
          </dependency>
          <dependency>
              <groupId>org.springframework</groupId>
              <artifactId>spring-tx</artifactId>
              <version>3.2.5.RELEASE</version>
          </dependency>
          <dependency>
              <groupId>org.springframework</groupId>
              <artifactId>spring-jdbc</artifactId>
              <version>3.2.5.RELEASE</version>
          </dependency>
          <dependency>
              <groupId>org.springframework.batch</groupId>
              <artifactId>spring-batch-core</artifactId>
              <version>3.0.0.M2</version>
          </dependency>
          <dependency>
              <groupId>org.springframework.batch</groupId>
              <artifactId>spring-batch-infrastructure</artifactId>
              <version>3.0.0.M2</version>
          </dependency>
          <dependency>
              <groupId>commons-dbcp</groupId>
              <artifactId>commons-dbcp</artifactId>
              <version>1.4</version>
          </dependency>
          <dependency>
              <groupId>javax.inject</groupId>
              <artifactId>javax.inject</artifactId>
              <version>1</version>
          </dependency>
          <dependency>
              <groupId>org.slf4j</groupId>
              <artifactId>jcl-over-slf4j</artifactId>
              <version>1.7.5</version>
          </dependency>
          <dependency>
              <groupId>org.slf4j</groupId>
              <artifactId>slf4j-api</artifactId>
              <version>1.7.5</version>
          </dependency>
          <dependency>
              <groupId>ch.qos.logback</groupId>
              <artifactId>logback-classic</artifactId>
              <version>1.0.13</version>
          </dependency>
          <dependency>
              <groupId>org.hsqldb</groupId>
              <artifactId>hsqldb</artifactId>
              <version>2.2.9</version>
          </dependency>
      </dependencies>
  </project>

EntryPointの作成

バッチ処理を書く前にバッチをキックするエントリポイントとなるmainクラスを作成します。

package com.example.jsr352;

import javax.batch.operations.JobOperator;
import javax.batch.runtime.BatchRuntime;

public class EntryPoint {

    public static void main(String[] args) {
        String jobId;
        if (args.length > 0) {
            jobId = args[0];
        } else {
            jobId = "hellojob";
        }

        JobOperator jobOperator = BatchRuntime.getJobOperator();
        System.out.println("start jobId=" + jobId);
        long executionId = jobOperator.start(jobId, null);
        System.out.println("executionId=" + executionId);
    }

}

JobOperatorを取得して、startメソッドに対象のジョブIDを渡すだけです。簡単。

最も簡単なBatchletの作成

最初はBatchlet方式のサンプルを説明します。

Batcheletクラス

package com.example.jsr352.job.hello;

import javax.batch.api.AbstractBatchlet;
import javax.inject.Named;

@Named
public class HelloBatchlet extends AbstractBatchlet {

    @Override
    public String process() throws Exception {
        System.out.println("hello world");
        return null;
    }

}

簡単ですね。processメソッドに処理を書き下すだけです。返り値は終了状態ですが、ここでは特に使用しないのでnullを返します。

ここまでSpringは現れません。

ジョブ定義ファイル

クラスパス以下ののMETA-INF/batch-jobs/[jobId].xmlが読み込まれます。ここではjobIdをhellojobとします。

  • META-INF/batch-jobs/hellojob.xml

      <?xml version="1.0" encoding="UTF-8"?>
  <beans xmlns="http://www.springframework.org/schema/beans"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xmlns:context="http://www.springframework.org/schema/context"
      xmlns:batch="http://www.springframework.org/schema/batch"
      xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch.xsd
          http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
          http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
  
      <context:component-scan base-package="com.example.jsr352.job.hello" />
  
      <job id="samplejob" xmlns="http://xmlns.jcp.org/xml/ns/javaee"
          version="1.0">
          <step id="step1">
              <batchlet ref="helloBatchlet"></batchlet>
          </step>
      </job>
  </beans>

    SpringのBean定義ファイルの中にJSR-352の<job>タグが含まれています。実は<job>タグだけでも使えるのですが、後々Springのサポートが必要になってくると思うのであえて初めからSpringの設定を入れています。ここでは@Namedによるコンポーネントスキャンを有効にするために<context:component-scan>を設定しておきます。

    <job>のなかに複数の<step>を定義できます。<step>の実現には、先ほど挙げた2パターンを選ぶことができます。ここでは<batchlet>を設定して、Batchlet方式を選びrefにBean IDを指定します。HelloBatchletに@NamedがついているのでデフォルトではhelloBatchletがBean IDになります。@Namedをつけない場合はFQCNを指定すればよいです。

先ほどのEntryPointの引数にjobIdとしてhellojobを渡して実行しましょう。

$ mvn -q exec:java -Dexec.mainClass=com.example.jsr352.EntryPoint -Dexec.arguments="hellojob"

start jobId=hellojob
hello world
executionId=0

簡単ですね。

次はDIをつかってロジック(?)を外だしてみます。

サービスクラス

package com.example.jsr352.job.hello;

import javax.inject.Named;

@Named
public class HelloService {    
    public String hello(String name) {
        return "hello " + name;
    }    	
}

Batchletクラス

package com.example.jsr352.job.hello;

import javax.batch.api.AbstractBatchlet;
import javax.inject.Inject;
import javax.inject.Named;

@Named
public class HelloBatchlet extends AbstractBatchlet {
    @Inject
    HelloService helloService;

    @Override
    public String process() throws Exception {
        System.out.println(helloService.hello("world"));
        return null;
    }

}

実行結果はさっきと同じです。このサンプルまでである程度出来そうなことが想像できるのではないでしょうか。ここまでプログラム中にSpringのパッケージは出てこないですね。

サービスのメソッドをトランザクション境界にしてみましょう。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:batch="http://www.springframework.org/schema/batch"
    xmlns:tx="http://www.springframework.org/schema/tx"
    xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch.xsd
        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
        http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">

    <context:component-scan base-package="com.example.jsr352.job.hello" />
    <tx:annotation-driven/>

    <job id="samplejob" xmlns="http://xmlns.jcp.org/xml/ns/javaee"
        version="1.0">
        <step id="step1">
            <batchlet ref="helloBatchlet2"></batchlet>
        </step>
    </job>
</beans>

<tx:annotation-driven />を追加します。

package com.example.jsr352.job.hello;

import javax.inject.Named;

import org.springframework.transaction.annotation.Transactional;

@Named
public class HelloService {
    @Transactional
    public String hello(String name) {
        return "hello " + name;
    }
}

サービスのメソッドまたはクラスに@Transactonalをつけます。ここでSpringのアノテーションがでてきました。現状SpringではJTA1.2に対応していないので、Springのアノテーションを使わざるをえないですが、Spring4になるとJTAの@Tranasctionalが使用できるはずなので、ここも将来的にはJava標準APIだけで書けるでしょう(別にSpringのアノテーションでも良いと思うが)。

ちなみにバッチ起動時にDataSourceやTransactionManagerが作られています。spring-batch-core-3.0.0.M2.jar内のbaseContext.xmlに色々定義されています。

ItemReader/ItemProcessor/ItemWriter方式の簡単なジョブを作成

次にETLパターンを試します。

  • Extract(抽出) -> ItemReader
  • Transform(変換) -> ItemProcessor
  • Load(書出) -> ItemWriter

が対応します。

簡単な例としてファイルの内容を読みこみ、偶数行は小文字に、奇数行は大文字の変換して、標準出力に書き込む例を実装してみます。

入力ファイル

  • input.txt

      aaa
  bbb
  ccc
  ddd
  eee
  fff
  ggg
  (中略)
  sss
  ttt
  uuu
  vvv
  www
  xxx
  yyy
  zzz

  • LineItemクラス

    ファイル行に対応したオブジェクトを作っておきます。

      package com.example.jsr352.job.file;
  
  public class LineItem {
  
      private final int lineNumber;
      private final String content;
  
      public LineItem(int lineNumber, String content) {
          this.lineNumber = lineNumber;
          this.content = content;
      }
  
      public int getLineNumber() {
          return lineNumber;
      }
  
      public String getContent() {
          return content;
      }
  
      @Override
      public String toString() {
          return "LineItem [lineNumber=" + lineNumber + ", content=" + content
                  + "]";
      }
  }

ItemReader

package com.example.jsr352.job.file;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.Serializable;

import javax.batch.api.chunk.AbstractItemReader;
import javax.inject.Named;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Named
public class FileItemReader extends AbstractItemReader {
    private static final Logger logger = LoggerFactory
            .getLogger(FileItemReader.class);

    BufferedReader reader;
    int lineNumber = 0;

    @Override
    public void open(Serializable checkpoint) throws Exception {
        logger.debug("open");
        reader = new BufferedReader(new FileReader("input.txt"));
    }

    @Override
    public Object readItem() throws Exception {
        String content = reader.readLine();
        return content == null ? null : new LineItem(++lineNumber, content);
    }

    @Override
    public void close() throws Exception {
        logger.debug("close");
        if (reader != null) {
            reader.close();
        }
    }

}

openでリソースを開き、readItemでリソースを任意の単位で読み込み行オブジェクトを作成します(StringでもOK)。closeでリソースを閉じます。

readItemの返り値がnullの場合に読み込みを終了します。

ItemProcessor

ItemReaderのreadItemで作成したオブジェクトがprocessItemに渡ってきます。Object型なのはきもいですが、気にしないでおきましょう。

package com.example.jsr352.job.file;

import javax.batch.api.chunk.ItemProcessor;
import javax.inject.Named;

@Named
public class FileItemProcessor implements ItemProcessor {
    @Override
    public Object processItem(Object item) throws Exception {
        LineItem lineItem = (LineItem) item;
        String content = lineItem.getContent();
        return (lineItem.getLineNumber() % 2 == 0) ? content.toLowerCase()
                : content.toUpperCase();
    }
}

ここでLineItemオブジェクトがStringに変換されます。

ItemWriter

ItemProcessorで変換されたオブジェクトがチャンク単位で渡ってきます。

package com.example.jsr352.job.file;

import java.util.List;

import javax.batch.api.chunk.AbstractItemWriter;
import javax.inject.Named;

@Named
public class FileItemWriter extends AbstractItemWriter {
    @Override
    public void writeItems(List<Object> items) throws Exception {
        for (Object item : items) {
            System.out.println(item);
        }
    }
}

List<Object>であることがミソですね。一行ずつ書き込むのではなく、まとまった単位(チャンク)で効率よく書き込めます。 デフォルトのチャンクサイズは10です。

ジョブ定義ファイル

  • META-INF/batch-jobs/filejob.xml

      <?xml version="1.0" encoding="UTF-8"?>
  <beans xmlns="http://www.springframework.org/schema/beans"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
      xmlns:batch="http://www.springframework.org/schema/batch"
      xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch.xsd
          http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
          http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
  
      <context:component-scan base-package="com.example.jsr352.job.file" />
  
      <job id="filejob" xmlns="http://xmlns.jcp.org/xml/ns/javaee"
          version="1.0">
          <step id="step1">
              <chunk>
                  <reader ref="fileItemReader" />
                  <processor ref="fileItemProcessor" />
                  <writer ref="fileItemWriter" />
              </chunk>
          </step>
      </job>
  </beans>

    Batchletの場合と大差ないですが、<chunk>の中身が<reader>、<processor>、<writer>になります。それぞれ先ほど作成したクラスのBean IDを指定します。

    チャンクサイズは<chunk item-count="20">のように設定できます。

実行してみます。

$ mvn -q exec:java -Dexec.mainClass=com.example.jsr352.EntrryPoint -Dexec.arguments="filejob"    
AAA
bbb
CCC
ddd
EEE
fff
GGG
(中略)
SSS
ttt
UUU
vvv
WWW
xxx
YYY
zzz

これも理解しやすいのではないでしょうか。

Spring Batchの機能を使ってみる

jBatchではItemReader、ItemProcessor、ItemWriterのAPIは規定していますが実装はユーザー任せです。

この概念はもともとSpring Batchのものであり、Spring Batchにはもちろん沢山の実装クラスが用意されています。

ここではDBからデータをカーソルで読み込むItemReaderを使ってみます。

ジョブ定義ファイル

まず始めにジョブ定義ファイルから見ていきます。

  • META-INF/batch-jobs/dbjob.xml

      <?xml version="1.0" encoding="UTF-8"?>
  <beans xmlns="http://www.springframework.org/schema/beans"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
      xmlns:batch="http://www.springframework.org/schema/batch"
      xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch.xsd
          http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
          http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
  
      <context:component-scan base-package="com.example.jsr352.job.db" />
  
      <bean id="dbReader"
          class="org.springframework.batch.item.database.JdbcCursorItemReader">
          <property name="sql" value="SELECT name FROM testdata" />
          <property name="rowMapper" ref="dbRowMapper" />
          <property name="dataSource" ref="dataSource" />
      </bean>
  
      <job id="dbjob" xmlns="http://xmlns.jcp.org/xml/ns/javaee"
          version="1.0">
          <step id="step1">
              <chunk>
                  <reader ref="dbReader" />
                  <writer ref="dbWriter" />
              </chunk>
          </step>
      </job>
  </beans>

    Readerは用意されているものを定義するだけです。Spring側でbean定義し、jBatch側で参照できます。

    SpringBatchで用意されているReaderはorg.springframework.batch.item.ItemReaderを実装しており、実はjBatchのjavax.batch.api.chunk.ItemReaderを実装していません。なぜ使えるかというと、内部ではjBatchのItemReaderは使っておらず、Spring BatchのItemReaderを使っており、jBatchのItemReaderはAdaptorを介してSpring BacthのItemReaderとして扱われるからです。

  • DbRowMapper

    JdbcCursorItemReaderが使用するRowMapperを作成します。ResultSetをオブジェクトに変換するクラスで、当然Springのクラスです。

      package com.example.jsr352.job.db;
  
  import java.sql.ResultSet;
  import java.sql.SQLException;
  
  import javax.inject.Named;
  
  import org.springframework.jdbc.core.RowMapper;
  
  @Named
  public class DbRowMapper implements RowMapper<String> {
      @Override
      public String mapRow(ResultSet rs, int rowNum) throws SQLException {
          return rs.getString(1);
      }
  }

ItemWriter

ItemProcessorは省略して、変換なしでItemWriterに渡します。ItemWriterも超手抜き。

package com.example.jsr352.job.db;

import java.util.List;

import javax.batch.api.chunk.AbstractItemWriter;
import javax.inject.Named;

@Named
public class DbWriter extends AbstractItemWriter {
    @Override
    public void writeItems(List<Object> items) throws Exception {
        System.out.println(items);
    }
}

DDL

Spring BatchはデフォルトでHSQLを使用し、起動時にクラスパスのorg/springframework/batch/core/schema-hsqldb.sqlを読み込みます。もちろん変更可能ですが、今回はここにスクリプトを書きます。超適当な例を貼っておきます。

CREATE TABLE testdata(name VARCHAR(10));
INSERT INTO testdata(name) VALUES('a');
INSERT INTO testdata(name) VALUES('b');
INSERT INTO testdata(name) VALUES('c');
(中略)
INSERT INTO testdata(name) VALUES('x');
INSERT INTO testdata(name) VALUES('y');
INSERT INTO testdata(name) VALUES('z');

実行してみます。

$ mvn -q exec:java -Dexec.mainClass=com.example.jsr352.EntrryPoint -Dexec.arguments="dbjob" 
start jobId=dbjob
[a, b, c, d, e, f, g, h, i, j]
[k, l, m, n, o, p, q, e, s, t]
[u, v, w, x, y, z]

チャンク数がわかりますね。

ということでjBatch(JSR-352) on Spring Batch3の基本的な使い方を説明してきましたが、

  • jBatchもSpring Batchの力を借りれば使えそう
  • Springはやはり基盤として優秀

ということがお分かりいただけたのではないでしょうか。

この記事のソースコードはこちら。

jBatchに関しては上妻さんのSlideShareが詳しいです。

またSpring Batch3連携としてはこのサンプルが参考になりました。

またの機会にエラーハンドリング、リスタート、リトライ、マルチスレッド処理等扱ってみたいと思います。

Spring Advent Calendarではあまり本には載っていないマニアックな内容を紹介していく予定なのでチェックよろしくです。

明日は@eiryuさんですね。よろしくです。

✒️️ Edit  ⏰ History  🗑 Delete