Java Springの逆引きメモ

・実際に作成するものとは（Stepの概念について）

・既に用意されている機能（クラス）を探すには？

・DBにデータを読み書きするには？

・データの書き込み（コミット）タイミングを制御するには？（restart機能もついでに見る！）

・バッチ処理を非同期で起動するには？

データの書き込み（コミット）タイミングを制御するには？（restartもついでに見る！）

SpringBatchには、コミット数を設定できる機能があると今までの記事で書いてきました。

ここでは、その機能についてみてみましょう！

ついでに、

処理の途中でエラーが発生したときにrestartさせた動きも見てみましょう。

ちなみに、

コミットはDBだけでなく、ファイルの書き込みにも関係していますので、

設定する数をうまく選んで処理スピードが速くなるようにしましょう

【サンプル】

サンプルは、前回の記事と同じものを使用します。

　　・DBにデータを読み書きするには？

　　上記の内容をすべて実行したら、以下の準備もお忘れなく！

　　　　・DBのmemberテーブルのデータをすべてクリアしておく

　　　　・起動の引数に、-nextをつけて実行する　（つけないとAlreadyCompleted例外が発生すると思います）

【説明】

再掲：

<job id="jobDb" xmlns="http://www.springframework.org/schema/batch "

　　incrementer="jobParametersIncrementer">

　　　<step id="step1" parent="simpleStep" >
　　　　　<tasklet >
　　　　　　　<chunk reader="fileItemReader" writer="dbItemWriter" commit-interval="2" />
　　　　　</tasklet>
　　</step>
</job>

　 commit-interval="2"となっています。

　つまり、コミットタイミングが2個と設定されています。

　データ2個ごとにデータがコミットされます。

　サンプルでは、ファイルからの読み込みになっているので、2行を溜め込んで、2行をDBに書き込んで

　コミットすると言うことになります。

　処理のイメージは以下のとおりです。

　　List items = new Arraylist();

　　for(int i = 0; i < commitInterval; i++){
　　　　items.add(itemReader.read());
　　}

　　itemWriter.write(items);

　　なんとなく動きが理解できましたでしょうか？

　　write()が内部で実際どのような処理になっているかは、使用するItemWriterによります。

　　ちなみにFlatFileItemWriterでも、commitIntervalの数を上げると格段に処理が早くなります。

　　しかし、数を上げすぎるとDBでもファイルでも、Out Of Memory例外が発生することもあります。

　　もちろん、起動時にheapサイズなどを設定しておけば対応できますが、限界があるので

　　最適値というものが存在します。

　　何回か走行して最適値を見つけましょう！

　　＜ページングを使用したDBの読み込みと書き込みのコミットについて＞

　　ページングItemReaderの機能を使用すると、一気に指定のページサイズ分だけデータを読み込みます。

　　そして、WriterがDBに値を書き込みます。

　　一気に操作する数の設定項目が、「ページサイズ」と「コミット数（commitInterval）」２つあります。

　　2つの設定が実際の動作にどのように関わるかを簡単に見てみます。

　　実際に実験した内容です。

　　　＜＜設定値＞＞

　　　ページサイズ：　1000

　　　コミット数　　：　　100

　　　＜＜動作＞＞

　　　１．select文が実行され、1000レコードが読み込まれる。

　　　２．insert文をバッチ機能で100レコード一気に挿入する。

　　　３．2を10回繰り返し、1000レコード分が処理される。

　　　４．１に戻って繰り返し。select文でデータ取れなくなるまで実行される。

　　なんとなく動作わかりましたでしょうか？

　　ちなみに、何回SQL文が実行されるでしょうか？

　　もし対象レコードが10000レコードあるとすると、

　　　　select文　＝　10000÷1000　＝　10回

　　　　insert文　＝　（10000÷1000）×（1000÷100）　＝　100回　

　　　　　　　　　（バッチ機能は1度SQL解釈してデータを一気に流し込みます）

　　ついでに、ページ機能、commitInterval機能を使用しなかった場合も見ておきましょう。

　　　select文　＝　１回

　　　insert文　＝　10000回

　　select文が少ない分、insert文の発行数が増えるので遅くなるのですね。

【restartの動作について】

では次に途中でエラーが発生した場合の動作を見てみましょう。

＜準備＞

前回の記事のサンプルでエラーを起こさせるには、以下のようにデータを書き替えます。

＜データ＞

1,太郎,18
2,次郎,29
3,さんま,53
abc,タモリ,54
5,鶴野,28

4行目が文字になっているので変換エラーが起きます。

しかも、コミット数は2にしているので、3行目のinsert処理をしてからエラーが起きます。

つまり、ちゃんとロールバックされていれば3行目の反映が無視され、2行目までが登録されているはずです。

＜実行結果＞

↓ログ

2010/08/21 11:39:50.312 [] INFO :: Job: [FlowJob: [name=jobDb]] launched with the following parameters: [{run.id=12}]
2010/08/21 11:39:50.937 [] INFO :: Executing step: [step1]
2010/08/21 11:39:51.156 [] ERROR :: Parsing error at line: 4 in resource=class path resource

　　　　　[sample/sample-db.txt], input=[abc,タモリ,54]
org.springframework.batch.item.file.FlatFileParseException: Parsing error at line: 4, input=[abc,タモリ,54]
at org.springframework.batch.item.file.mapping.DefaultLineMapper.mapLine(DefaultLineMapper.java:49)
at org.springframework.batch.item.file.FlatFileItemReader.doRead(FlatFileItemReader.java:188)
at org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader.read(AbstractItemCountingItemStreamItemReader.java:85)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
at java.lang.reflect.Method.invoke(Unknown Source)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:307)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:182)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:149)
（中略）
Caused by: java.lang.NumberFormatException: Unparseable number: abc
at org.springframework.batch.item.file.transform.DefaultFieldSet.parseNumber(DefaultFieldSet.java:688)
at org.springframework.batch.item.file.transform.DefaultFieldSet.readInt(DefaultFieldSet.java:290)
at sample.SimpleMapFieldSetMapper.read(SimpleMapFieldSetMapper.java:42)
at sample.SimpleMapFieldSetMapper.mapFieldSet(SimpleMapFieldSetMapper.java:34)
at sample.SimpleMapFieldSetMapper.mapFieldSet(SimpleMapFieldSetMapper.java:1)
at org.springframework.batch.item.file.mapping.DefaultLineMapper.mapLine(DefaultLineMapper.java:46)
... 43 more
2010/08/21 11:39:51.234 [] INFO :: Job: [FlowJob: [name=jobDb]] completed with the following parameters: [{run.id=12}] and the following status: [FAILED]

＜さらにrestartする＞

さて、予定通りエラーになりました。

ログにはちゃんとエラーになった行番号とデータの内容が出力されています！

DBのテーブルの中身を確認してみてください。2行目までしか登録されてないですよね！

では、restartしてみましょう。

まず、ファイルのデータを元に戻してください。（これ忘れずに！）

1,太郎,18
2,次郎,29
3,さんま,53
4,タモリ,54
5,鶴野,28

次にコマンドの実行するときに引数タブ内の記述で、-restartをつけて実行してください。

　　classpath:/sample/runDb-context.xml jobDb -restart

＜処理結果＞

↓ログ

2010/08/21 11:48:29.078 [] INFO :: Job: [FlowJob: [name=jobDb]] launched with the following parameters: [{run.id=12}]
2010/08/21 11:48:29.296 [] INFO :: Executing step: [step1]
2010/08/21 11:48:29.468 [] INFO :: Job: [FlowJob: [name=jobDb]]

　　　　completed with the following parameters: [{run.id=12}] and the following status: [COMPLETED]

正常に終了しましたね。

DBのテーブルも見てみてください。

残りのデータが全部登録されました！

【まとめ】

SpringBatchを使用するだけで、手間はSQLのコマンドシェルみたいなものを作るのと同じくらいで

コミットタイミングの制御、およびrestartの機能まで実装できることを見ました。

かなり便利です。

少し覚えることはありますが、直感に近いのでそれほど惑わないと思います。

同様のSpringを使用したバッチ機能にTERAバッチというフレームワークがありますが

こちらはどうなんでしょうね。

気になります。

参照：

・トップ

・SpringBatch機能について

・CommandLineJobRunnerとは？

・Executionとは？　（SpringBatch用語）

・WEB上で定期的に何か処理をするには？

・DBにデータを読み書きするには？

DBにデータを読み書きするには？

ここまでの記事で、SpringBatchのバッチ処理の記述の仕方を見てきました。

　　・Spring Batchを使えるようにするには？（準備編）

しかし、ファイルの読み書きばかりでした。

この記事ではDBを扱うサンプルを見てみようと思います。

【処理内容】

まず、これからどのようなことをやろうとしているかを書いておきます。

　①ファイルからデータを読み込みます。

　②DBのテーブルに読み込んだデータを書き込みます。

【準備】

SpringBatchの準備は今までと同じです。以下の記事を参考に準備をお願いします！

　　※DBも作成してください。（サンプルのdataSourceはPostgresになっています）

さらに、今回はデータ書き込み先のテーブルも作っておきます。

＜テーブル　（使用するDBはPostgeresです）＞

CREATE TABLE member
(
id integer NOT NULL,
uname varchar(20),
age integer,
in_date date
)

　※他のDBを使用する場合は、DBに合わせてcreate文、dataSourceの設定内容を変更してください。

　＜パッケージの作成＞
　プロジェクトには/sample　パッケージを作成しておいてください。

【 /sample/back-context.xml ファイル（Batchバックグラウンドの設定）】

　以下の記事と同じ物を使いまわします。

　リンク内の同じ項目の内容をファイルにしてください。

【SimpleMapFieldSetMapperクラス】
　これも前回の記事を使いまわしますので、同じ項目の内容をコピーしてください。

【 /sample/sample-db.txt ファイル（読み込みファイル）】

1,太郎,18
2,次郎,29
3,さんま,53
4,タモリ,54
5,鶴野,28

【 /sample/runDb-context.xml ファイル（バッチ処理の設定）】

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" 
 xmlns:p="http://www.springframework.org/schema/p" 
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="http://www.springframework.org/schema/batch
     http://www.springframework.org/schema/batch/spring-batch-2.1.xsd
     http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
 ">
 <import resource="classpath:/sample/back-context.xml"/>

 <!-- ジョブの処理 -->
 <job id="jobDb" xmlns="http://www.springframework.org/schema/batch
 " incrementer="jobParametersIncrementer">
 
  <step id="step1" parent="simpleStep" >
   <tasklet >
    <chunk reader="fileItemReader" writer="dbItemWriter" commit-interval="2" />
   </tasklet>
  </step>
  
 </job>
 
 
 
 <!-- enables the functionality of JobOperator.startNextInstance(jobName) -->
 <bean id="jobParametersIncrementer" class="org.springframework.batch.core.launch.support.RunIdIncrementer" />
 <bean id="simpleStep"
  class="org.springframework.batch.core.step.item.SimpleStepFactoryBean"
  abstract="true">
  <property name="jobRepository" ref="jobRepository" />
  <property name="startLimit" value="100" />
  <property name="commitInterval" value="1" />
 </bean>


 <!-- ファイルを読み込みMapに入れる -->
 <bean id="fileItemReader" class="org.springframework.batch.item.file.FlatFileItemReader"  scope="step">
  <property name="resource" value="classpath:/sample/sample-db.txt" />
  <property name="encoding" value="shift_jis" />
  <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="id,uname,age" />
     </bean>
    </property>
    <property name="fieldSetMapper">
     <bean class="sample.SimpleMapFieldSetMapper" >
      <property name="types" value="int,String,int" />
     </bean>
    </property>
   </bean>
  </property>
 </bean>


 <!-- SQLを実行していく（readしたMapをプレースホルダに設定して実行する） 
 -->
 <bean id="dbItemWriter" class="org.springframework.batch.item.database.JdbcBatchItemWriter">
  <property name="dataSource" ref="dataSource" />
  <property name="itemSqlParameterSourceProvider">
   <bean class="sample.MapSqlParameterSourceProvider" />
  </property>
  <property name="itemPreparedStatementSetter" >
   <bean class="org.springframework.batch.item.database.support.ColumnMapItemPreparedStatementSetter" />
  </property>
  <property name="sql" value="
   insert into member(id, uname, age)
   values(:id, :uname, :age)
  " />
 </bean>
 
 
</beans>







【 /sample/MapSqlParameterSourceProvider （SpringJdbc関連のクラス）】

package sample;
import java.util.Map;
import org.springframework.batch.item.database.ItemSqlParameterSourceProvider;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;

public class MapSqlParameterSourceProvider implements
  ItemSqlParameterSourceProvider<Map<String, Object>> {
 @Override
 public SqlParameterSource createSqlParameterSource(Map<String, Object> map) {
  return new MapSqlParameterSource(map);
 }
}

【起動方法】

前回までと同じです。

以下のリンクの【実行】と同じことをしてみてください。

ただし、「プログラムの引数」はちょっと変えます。

・実際に作成するものとは（Stepの概念について）

　　プログラムの引数　：　classpath:/sample/runDb-context.xml jobDb

　　※ただし、2回目以降は状態がCOMPLETEDになっているため実行エラーが出ます。

　　　この場合、プログラムの引数に　-nextを付け加えてみてください。

　　　（参考：・CommandLineJobRunnerとは？）

　　　あと、重複キー違反も起きると思うのでテーブルのデータも一度クリアしてください。

【結果】

2010/08/19 23:29:03.921 [] INFO :: Job: [FlowJob: [name=jobDb]]

　　　　　　　　launched with the following parameters: [{run.id=1}]
2010/08/19 23:29:04.062 [] INFO :: Executing step: [step1]
2010/08/19 23:29:04.265 [] INFO :: Job: [FlowJob: [name=jobDb]]

　　　　　　　　completed with the following parameters: [{run.id=1}] and the following status: [COMPLETED]

【説明】

どうでしょうか？

無事、DBにデータが書き込まれましたでしょうか？

やっていることは今までの記事の内容と同じです。

①ItemReaderでファイルからデータを読み込みます。
　　これは前回の記事、そのままです。

　　ですので、説明は前回の記事に譲ります。

　　　・Stepを使って処理を書くには？　　（作成したクラスSimpleMapFieldSetMapperについて）

②ItemWriterで読み込んだデータを書き込みます。

　JdbcBatchItemWriterを使用しています。

再掲：

　</property>
　<property name="itemPreparedStatementSetter" >
　　　<bean class="org.springframework.batch.item.database.support.ColumnMapItemPreparedStatementSetter" />
　</property>
　<property name="sql" value="
　　　insert into member(id, uname, age)
　　　values(:id, :uname, :age)
　" />
</bean>

　　JdbcBatchItemWriterは、SpringJdbcのバッチ機能を使用してDBにアクセスします。

　　SpringJdbcの詳細は、・SpringJDBCの機能についてを参照してください。

　　SpringJdbcは、"？"をプレースホルダにするタイプと、":名前"をプレースホルダにするタイプがあります。

　　上記では、後者を使用しています。（sqlプロパティを見てみてください。）

　　itemSqlParameterSourceProviderは、プレースホルダに使用する値をどのように与えるかを

　　定義するプロパティです。

　　ItemReaderは、LinkedHashMapを返すように設定したので、Mapのキー名をプレースホルダに

　　展開したいのですが、そのようなクラスは既存のSpringBatchでは用意されていません。

　　ですので、自作しました。（MapSqlParameterSourceProvider）

　　SpringJdbcのクラスである、MapSqlParameterSourceにMapを渡すだけの単純なクラスです。

　　本当はMapではなく、Beanを使用すると自作する必要が無いのですが、実験をするときは

　　やはりメンドウなので、こんなクラスを作っておくと使いまわせますし便利ですよね！

　　さて、この例で分かったかと思いますが、Spring BatchはSpringJdbcの機能を簡単に扱えます。

　　バッチ処理をDBで扱う場合、たいていはSQL文の羅列をコマンドでDBに流すことになると思います。

　　それはそれで分かりやすいし便利ですが、正常に終わらなかったときは事です。

　　エラーなどで途中で処理が終わった場合、どこまで終わっているのか？、どのような復旧処理をしてから

　　再実行すればいいのか？、途中から実行できないのか？、など問題点にいとまがありません

　　SpringBatchでは、処理がどのstepまで終わっているか？などをDBに保存しています。

　　restartすればDBの値を参考に前回停止したところから処理を再開してくれます。

　　コマンドでDBに流す場合と同じく、SQL文を書くだけのところは変わらないのに、これだけ違うのです。

　　すごく便利だと思いません？

【JdbcBatchItemWriterについて】

クラス名に、Batchという言葉が入っています。

Batchというのは、指定の回数だけ一気にデータをDBに送り込むことを示しています。

標準のJavaのJDBCのクラスにもBatchというものがあるので、一般的な機能のようです。

データを一気に送り込むことで、１回１回送り込むよりもDBの処理が早くなります。

バッチ処理を行うときは大量データを送ることが多いと思います。

かなり重宝します！

また、先ほどコマンドでDBに流し込む話をしましたが、その弱点はもう一つあります。

指定の回数だけ一気にデータを送り込んだり、トランザクションのコミットタイミングを制御するのが難しいことです。

SpringBatchを使用すれば、コミットタイミングについても設定だけで可能になります。

コミットタイミングについては別の記事で見てみようと思います。

【その他のORマッピングフレームワーク】

上記ではSpringJDBCを利用できることがわかりました。

では、他のORマッピングについてはどうでしょうか？

他にも使えるものがあります。以下のものです。

　　・IBatis

　　・Hibernate

JavaDocを確認してみてください。

また、使えなくても使えるように仲介者クラスを作っておけば大丈夫かと思います。

ここではItemWrtiterしか紹介しませんでしたが、もちろんItemReaderも用意されています。

特に、ページングと言う機能は便利ですので以下で簡単に紹介しておきます。

【ItemReaderのページング機能とは】

簡単にページングについて紹介します。

SpringBatchではページング以外のDB読み込みの機能もありますが、ページングの機能だけで十分ではないかなと思っています。

まあ、とにかく見てみましょう。

例題ではIBatisを使用しますが、他のものでも考え方は同じです。

　iBatisのSQL文設定ファイル：

　　<select id="getPagingMember" resultClass="java.util.HashMap">
　　　 select
　　　　　　id,
　　　　　　name,
　　　　　　age
　　　　from member

　　　　order by id asc
　　　　offset #_skiprows# LIMIT #_pagesize#
　　</select>

　説明：

　　ページングとは、WEBのページングのように、ページ番号と１ページのデータ数を指定した検索です。

　　ページ番号は０～です。

　　IbatisPagingItemReaderは、#_skiprows#, #_pagesize#さらに、#_page#というパラメタを

　　自動で設定してくれます。

　　これらをうまく組み合わせてSQL文で返却するレコードを調整します。

　　そのPostgresの例題が、上記のSQL文です。

　　参考までにOracleの例も書いておきます。

　　Oracleの場合、limitが使えない分やっかいです。

　　　select
　　　　id,
　　　　name,
　　　　age
　　　from (
　　　　SELECT row_number() over(order by id) rn,
　　　　　　id,
　　　　　　name,
　　　　　　age
　　　　from member
　　　)
　　　where

　　　　rn > (#_page# * #_pagesize#)
　　　　and rn <= ( (#_page# + 1) * #_pagesize# )

　　　order by id

　　　　※本家SpringBatchのドキュメントでは、ROW_NUMを使用する方法がかかれていましたが

　　　　　上記のようにwindow関数を使用する方がきれいかなぁと個人的に思っています。

　　補足：

　　　ページングは、順番が変わらないフィールドをorder byの対象にすべきだと思います。

　　　処理途中から再実行するときに、再実行までに誰かがデータを増やした場合、順番が変わってしまうと

　　　次に処理するデータが狂ってしまうからです。

　　　たいていの場合、一意のフィールドやシーケンス番号を入れているフィールドなどがあるかと思います。

　　　それらのフィールドを組み合わせれば順番が変わらないようなSQL文も頑張れば作れるでしょう。

さて、大まかですがDBの読み書きについてみてきましたが、分かりましたでしょうか？

いつも分かりやすいか？と自問しながら記事を書いているので、理解できましたらうれしいです

参照：

・トップ

・SpringBatch機能について

・データの書き込み（コミット）タイミングを制御するには？（restart機能もついでに見る！）

・CommandLineJobRunnerとは？