ItemReader および ItemWriter
すべてのバッチ処理は、大量のデータの読み取り、ある種の計算または変換の実行、結果の書き込みとして、最も単純な形式で説明できます。Spring Batch は、一括読み取りおよび書き込みの実行に役立つ 3 つの主要なインターフェースを提供します: ItemReader、ItemProcessor、ItemWriter。
ItemReader
シンプルなコンセプトですが、ItemReader は多くの異なる型の入力からデータを提供する手段です。最も一般的な例は次のとおりです。
フラットファイル: フラットファイルアイテムリーダーは、通常ファイル内の固定位置で定義されたデータフィールドまたは特殊文字（コンマなど）で区切られたデータフィールドを持つレコードを記述するフラットファイルからデータ行を読み取ります。
XML: XML
ItemReaderは、オブジェクトの解析、マッピング、検証に使用されるテクノロジーとは独立して XML を処理します。入力データにより、XSD スキーマに対する XML ファイルの検証が可能になります。データベース: データベースリソースにアクセスして、処理のためにオブジェクトにマッピングできる結果セットを返します。デフォルトの SQL
ItemReader実装は、RowMapperを呼び出してオブジェクトを返し、再起動が必要な場合は現在の行を追跡し、基本的な統計を保存し、後で説明するトランザクションの強化を提供します。
さらに多くの可能性がありますが、この章の基本的なものに焦点を当てます。利用可能なすべての ItemReader 実装の完全なリストは、付録 A にあります。
ItemReader は、次のインターフェース定義に示すように、一般的な入力操作の基本的なインターフェースです。
public interface ItemReader<T> {
T read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException;
}read メソッドは、ItemReader の最も重要な契約を定義します。それを呼び出すと、1 つのアイテム、またはアイテムが残っていない場合は null が返されます。アイテムは、ファイル内の行、データベース内の行、XML ファイル内の要素を表す場合があります。通常、これらは使用可能なドメインオブジェクト（Trade、Foo など）にマップされることが期待されますが、契約ではそうする必要はありません。
ItemReader インターフェースの実装はフォワードのみであることが期待されます。ただし、基礎となるリソースがトランザクション（JMS キューなど）の場合、read を呼び出すと、ロールバックシナリオの後続の呼び出しで同じ論理項目が返される場合があります。また、ItemReader で処理するアイテムが不足していても、例外がスローされることはありません。例: 0 の結果を返すクエリで構成されたデータベース ItemReader は、read の最初の呼び出しで null を返します。
ItemWriter
ItemWriter の機能は ItemReader に似ていますが、逆の操作があります。リソースは引き続き検索、オープン、クローズする必要がありますが、ItemWriter が読み込むのではなく書き込むという点で異なります。データベースまたはキューの場合、これらの操作は挿入、更新、送信になります。出力の直列化の形式は、各バッチジョブに固有です。
ItemReader と同様に、ItemWriter は、次のインターフェース定義に示すように、かなり汎用的なインターフェースです。
public interface ItemWriter<T> {
void write(List<? extends T> items) throws Exception;
}ItemReader 上の read と同様に、write は ItemWriter の基本契約を提供します。開いている限り、渡されたアイテムのリストを書き込もうとします。通常、アイテムはまとめて「バッチ」にまとめられてから出力されることが予想されるため、インターフェースはアイテム自体ではなくアイテムのリストを受け入れます。リストを書き出した後、書き込みメソッドから戻る前に、必要なフラッシュを実行できます。例: Hibernate DAO に書き込む場合、各項目に 1 つずつ、複数の書き込み呼び出しを行うことができます。ライターは、戻る前に休止状態セッションで flush を呼び出すことができます。
ItemStream
ItemReader と ItemWriter は両方ともそれぞれの目的を十分に果たしますが、別のインターフェースを必要とする共通の懸念が両方にあります。一般に、バッチジョブの範囲の一部として、リーダーとライターを開いて閉じ、状態を保持するメカニズムが必要です。次の例に示すように、ItemStream インターフェースはその目的を果たします。
public interface ItemStream {
void open(ExecutionContext executionContext) throws ItemStreamException;
void update(ExecutionContext executionContext) throws ItemStreamException;
void close() throws ItemStreamException;
} 各メソッドを説明する前に、ExecutionContext にメンションする必要があります。ItemStream も実装する ItemReader のクライアントは、read を呼び出す前に open を呼び出して、ファイルなどのリソースを開いたり、接続を取得したりする必要があります。同様の制限が、ItemStream を実装する ItemWriter に適用されます。第 2 章で記述されていたように、予想されるデータが ExecutionContext で見つかった場合、それを使用して ItemReader または ItemWriter を初期状態以外の場所で開始することができます。逆に、close は、オープン中に割り当てられたリソースが安全に解放されるようにするために呼び出されます。update は主に、現在保持されている状態が提供された ExecutionContext にロードされることを保証するために呼び出されます。このメソッドは、コミット前に呼び出され、コミット前に現在の状態がデータベースに保持されるようにします。
ItemStream のクライアントが Step （Spring Batch コアから）である特別な場合、StepExecution ごとに ExecutionContext が作成され、同じ JobInstance が返された場合にユーザーが特定の実行の状態を保存できるようになります。再び開始されます。Quartz に精通している人にとって、セマンティクスは Quartz JobDataMap と非常に似ています。
デリゲートパターンとステップへの登録
CompositeItemWriter は、Spring Batch で一般的な委譲パターンの例であることに注意してください。デリゲート自体が、StepListener などのコールバックインターフェースを実装する場合があります。もしそうなら、それらが Job の Step の一部として Spring Batch コアと組み合わせて使用されているなら、それらはほぼ確実に Step に手動で登録する必要があります。ItemStream または StepListener インターフェースを実装している場合、Step に直接接続されているリーダー、ライター、プロセッサーは自動的に登録されます。ただし、デリゲートは Step に認識されていないため、リスナーまたはストリーム（または適切な場合は両方）として注入する必要があります。
次の例は、XML のストリームとしてデリゲートを挿入する方法を示しています。
<job id="ioSampleJob">
<step name="step1">
<tasklet>
<chunk reader="fooReader" processor="fooProcessor" writer="compositeItemWriter"
commit-interval="2">
<streams>
<stream ref="barWriter" />
</streams>
</chunk>
</tasklet>
</step>
</job>
<bean id="compositeItemWriter" class="...CustomCompositeItemWriter">
<property name="delegate" ref="barWriter" />
</bean>
<bean id="barWriter" class="...BarWriter" />次の例は、XML のストリームとしてデリゲートを挿入する方法を示しています。
@Bean
public Job ioSampleJob() {
return this.jobBuilderFactory.get("ioSampleJob")
.start(step1())
.build();
}
@Bean
public Step step1() {
return this.stepBuilderFactory.get("step1")
.<String, String>chunk(2)
.reader(fooReader())
.processor(fooProcessor())
.writer(compositeItemWriter())
.stream(barWriter())
.build();
}
@Bean
public CustomCompositeItemWriter compositeItemWriter() {
CustomCompositeItemWriter writer = new CustomCompositeItemWriter();
writer.setDelegate(barWriter());
return writer;
}
@Bean
public BarWriter barWriter() {
return new BarWriter();
}フラットファイル
バルクデータを交換するための最も一般的なメカニズムの 1 つは、常にフラットファイルです。XML が構造化方法（XSD）を定義するための合意された標準とは異なり、フラットファイルを読む人はファイルの構造化方法を正確に事前に理解する必要があります。一般に、すべてのフラットファイルは、区切りと固定長の 2 つの型に分類されます。区切りファイルは、フィールドがコンマなどの区切り文字で区切られているファイルです。固定長ファイルには、設定された長さのフィールドがあります。
FieldSet
Spring Batch でフラットファイルを操作する場合、入力用か出力用かにかかわらず、最も重要なクラスの 1 つは FieldSet です。多くのアーキテクチャとライブラリには、ファイルからの読み取りを支援する抽象化が含まれていますが、通常は String または String オブジェクトの配列を返します。これは本当に途中であなたを得るだけです。FieldSet は、ファイルリソースのフィールドをバインドできるようにするための Spring Batch の抽象化です。開発者は、データベース入力を操作するのとほぼ同じ方法でファイル入力を操作できます。FieldSet は、概念的に JDBC ResultSet に似ています。FieldSet には、String トークンの配列という 1 つの引数のみが必要です。オプションで、次の例に示すように、ResultSet の後にパターン化されたインデックスまたは名前でフィールドにアクセスできるように、フィールドの名前を構成することもできます。
String[] tokens = new String[]{"foo", "1", "true"};
FieldSet fs = new DefaultFieldSet(tokens);
String name = fs.readString(0);
int value = fs.readInt(1);
boolean booleanValue = fs.readBoolean(2);Date、long、BigDecimal など、FieldSet インターフェースにはさらに多くのオプションがあります。FieldSet の最大の利点は、フラットファイル入力の一貫した解析を提供することです。予期せぬ方法で各バッチジョブが異なる方法で解析するのではなく、フォーマット例外に起因するエラーを処理するとき、または単純なデータ変換を行うときの両方で一貫性があります。
FlatFileItemReader
フラットファイルは、最大 2 次元（表）データを含む任意の型のファイルです。Spring Batch フレームワークでのフラットファイルの読み取りは、FlatFileItemReader と呼ばれるクラスによって促進されます。このクラスは、フラットファイルの読み取りと解析のための基本的な機能を提供します。FlatFileItemReader の 2 つの最も重要な必須依存関係は、Resource と LineMapper です。LineMapper インターフェースについては、次のセクションで詳しく説明します。リソースプロパティは Spring Core Resource を表します。この型の Bean の作成方法を説明するドキュメントは、Spring Framework、第 5 章。リソースにあります。このガイドでは、次の簡単な例を示す以外に、Resource オブジェクトの作成の詳細には触れません。
Resource resource = new FileSystemResource("resources/trades.csv");複雑なバッチ環境では、ディレクトリ構造はエンタープライズアプリケーション統合（EAI）インフラストラクチャによって管理されることが多く、FTP の場所からバッチ処理の場所へ、その逆にファイルを移動するための外部インターフェースのドロップゾーンが確立されます。ファイル移動ユーティリティは Spring Batch アーキテクチャの範囲を超えていますが、バッチジョブストリームがファイル移動ユーティリティをジョブストリームのステップとして含むことは珍しいことではありません。バッチアーキテクチャに必要なのは、処理するファイルを見つける方法だけです。Spring Batch は、この開始点からパイプにデータを供給するプロセスを開始します。ただし、Spring Integration はこれらの型のサービスの多くを提供します。
FlatFileItemReader の他のプロパティでは、次の表で説明するように、データの解釈方法をさらに指定できます。
| プロパティ | タイプ | 説明 |
|---|---|---|
comments | String[] | コメント行を示す行プレフィックスを指定します。 |
encoding | String | 使用するテキストエンコーディングを指定します。デフォルトは |
lineMapper |
|
|
linesToSkip | int | ファイルの先頭で無視する行数。 |
recordSeparatorPolicy | RecordSeparatorPolicy | 行末がどこにあるかを判断し、引用符で囲まれた文字列内にある場合、行末を越えて続行するなどの操作を行うために使用されます。 |
resource |
| 読み取り元のリソース。 |
skippedLinesCallback | LineCallbackHandler | スキップするファイルの行の生の行コンテンツを渡すインターフェース。 |
strict | boolean | 厳格モードでは、入力リソースが存在しない場合、リーダーは |
LineMapper
ResultSet などの低レベルの構造体を取り、Object を返す RowMapper と同様に、フラットファイル処理では、String 行を Object に変換するために同じ構造体が必要です。次のインターフェース定義に示します。
public interface LineMapper<T> {
T mapLine(String line, int lineNumber) throws Exception;
} 基本的な契約は、現在の行とそれが関連付けられている行番号を指定すると、マッパーが結果のドメインオブジェクトを返すことです。これは、ResultSet の各行が行番号に結び付けられているように、各行がその行番号に関連付けられているという点で、RowMapper に似ています。これにより、結果のドメインオブジェクトに行番号を結び付けて、ID の比較やより詳細なログを記録できます。ただし、RowMapper とは異なり、LineMapper には生の行が与えられ、上記で説明したように、途中までしか行けません。このドキュメントで後述するように、行を FieldSet にトークン化してから、オブジェクトにマッピングする必要があります。
LineTokenizer
入力の行を FieldSet に変換するための抽象化が必要です。これは、FieldSet に変換する必要のあるフラットファイルデータの形式が多数存在する可能性があるためです。Spring Batch では、このインターフェースは LineTokenizer です。
public interface LineTokenizer {
FieldSet tokenize(String line);
}LineTokenizer の契約は、入力の行（理論的には String が複数の行を含むことができる）が与えられると、その行を表す FieldSet が返されるようなものです。次に、この FieldSet を FieldSetMapper に渡すことができます。Spring Batch には、次の LineTokenizer 実装が含まれています。
DelimitedLineTokenizer: レコード内のフィールドが区切り文字で区切られているファイルに使用されます。最も一般的な区切り文字はコンマですが、パイプまたはセミコロンもよく使用されます。FixedLengthTokenizer: レコード内のフィールドがそれぞれ「固定幅」であるファイルに使用されます。各フィールドの幅は、レコード型ごとに定義する必要があります。PatternMatchingCompositeLineTokenizer: パターンと照合することにより、特定の行でトークナイザーのリストのどのLineTokenizerを使用するかを決定します。
FieldSetMapper
FieldSetMapper インターフェースは、FieldSet オブジェクトを取り、その内容をオブジェクトにマップする単一のメソッド mapFieldSet を定義します。このオブジェクトは、ジョブのニーズに応じて、カスタム DTO、ドメインオブジェクト、配列になります。FieldSetMapper は LineTokenizer と組み合わせて使用され、次のインターフェース定義に示すように、リソースからのデータ行を目的の型のオブジェクトに変換します。
public interface FieldSetMapper<T> {
T mapFieldSet(FieldSet fieldSet) throws BindException;
} 使用されるパターンは、JdbcTemplate が使用する RowMapper と同じです。
DefaultLineMapper
フラットファイルを読み取るための基本的なインターフェースが定義されたため、次の 3 つの基本的な手順が必要であることが明らかになります。
ファイルから 1 行読み取ります。
String行をLineTokenizer#tokenize()メソッドに渡して、FieldSetを取得します。トークン化から返された
FieldSetをFieldSetMapperに渡し、ItemReader#read()メソッドから結果を返します。
上記の 2 つのインターフェースは、2 つの個別のタスクを表します。行を FieldSet に変換し、FieldSet をドメインオブジェクトにマッピングします。LineTokenizer の入力は LineMapper （ライン）の入力と一致し、FieldSetMapper の出力は LineMapper の出力と一致するため、LineTokenizer と FieldSetMapper の両方を使用するデフォルトの実装が提供されます。次のクラス定義に示されている DefaultLineMapper は、ほとんどのユーザーが必要とする動作を表しています。
public class DefaultLineMapper<T> implements LineMapper<>, InitializingBean {
private LineTokenizer tokenizer;
private FieldSetMapper<T> fieldSetMapper;
public T mapLine(String line, int lineNumber) throws Exception {
return fieldSetMapper.mapFieldSet(tokenizer.tokenize(line));
}
public void setLineTokenizer(LineTokenizer tokenizer) {
this.tokenizer = tokenizer;
}
public void setFieldSetMapper(FieldSetMapper<T> fieldSetMapper) {
this.fieldSetMapper = fieldSetMapper;
}
}上記の機能は、リーダーフレームワークの以前のバージョンで行われていたように、リーダー自体に組み込まれるのではなく、デフォルトの実装で提供され、特に生の行へのアクセスが必要な場合に、解析プロセスをより柔軟に制御できるようにします。
単純な区切りファイルの読み取りの例
次の例は、実際のドメインシナリオでフラットファイルを読み取る方法を示しています。この特定のバッチジョブは、次のファイルからフットボール選手を読み取ります。
ID,lastName,firstName,position,birthYear,debutYear "AbduKa00,Abdul-Jabbar,Karim,rb,1974,1996", "AbduRa00,Abdullah,Rabih,rb,1975,1999", "AberWa00,Abercrombie,Walter,rb,1959,1982", "AbraDa00,Abramowicz,Danny,wr,1945,1967", "AdamBo00,Adams,Bob,te,1946,1969", "AdamCh00,Adams,Charlie,wr,1979,2003"
このファイルの内容は、次の Player ドメインオブジェクトにマップされます。
public class Player implements Serializable {
private String ID;
private String lastName;
private String firstName;
private String position;
private int birthYear;
private int debutYear;
public String toString() {
return "PLAYER:ID=" + ID + ",Last Name=" + lastName +
",First Name=" + firstName + ",Position=" + position +
",Birth Year=" + birthYear + ",DebutYear=" +
debutYear;
}
// setters and getters...
}FieldSet を Player オブジェクトにマップするには、次の例に示すように、プレーヤーを返す FieldSetMapper を定義する必要があります。
protected static class PlayerFieldSetMapper implements FieldSetMapper<Player> {
public Player mapFieldSet(FieldSet fieldSet) {
Player player = new Player();
player.setID(fieldSet.readString(0));
player.setLastName(fieldSet.readString(1));
player.setFirstName(fieldSet.readString(2));
player.setPosition(fieldSet.readString(3));
player.setBirthYear(fieldSet.readInt(4));
player.setDebutYear(fieldSet.readInt(5));
return player;
}
} 次に、次の例に示すように、FlatFileItemReader を正しく構築し、read を呼び出すことにより、ファイルを読み取ることができます。
FlatFileItemReader<Player> itemReader = new FlatFileItemReader<>();
itemReader.setResource(new FileSystemResource("resources/players.csv"));
DefaultLineMapper<Player> lineMapper = new DefaultLineMapper<>();
//DelimitedLineTokenizer defaults to comma as its delimiter
lineMapper.setLineTokenizer(new DelimitedLineTokenizer());
lineMapper.setFieldSetMapper(new PlayerFieldSetMapper());
itemReader.setLineMapper(lineMapper);
itemReader.open(new ExecutionContext());
Player player = itemReader.read();read を呼び出すたびに、ファイルの各行から新しい Player オブジェクトが返されます。ファイルの終わりに達すると、null が返されます。
名前によるフィールドのマッピング
DelimitedLineTokenizer と FixedLengthTokenizer の両方で許可され、機能が JDBC ResultSet に似ている追加機能が 1 つあります。フィールドの名前をこれらの LineTokenizer 実装のいずれかに挿入して、マッピング機能の可読性を高めることができます。最初に、次の例に示すように、フラットファイル内のすべてのフィールドの列名がトークナイザーに挿入されます。
tokenizer.setNames(new String[] {"ID", "lastName", "firstName", "position", "birthYear", "debutYear"});FieldSetMapper は、この情報を次のように使用できます。
public class PlayerMapper implements FieldSetMapper<Player> {
public Player mapFieldSet(FieldSet fs) {
if (fs == null) {
return null;
}
Player player = new Player();
player.setID(fs.readString("ID"));
player.setLastName(fs.readString("lastName"));
player.setFirstName(fs.readString("firstName"));
player.setPosition(fs.readString("position"));
player.setDebutYear(fs.readInt("debutYear"));
player.setBirthYear(fs.readInt("birthYear"));
return player;
}
}FieldSets をドメインオブジェクトに自動マッピングする
多くの人にとって、特定の FieldSetMapper を作成する必要があることは、JdbcTemplate に対して特定の RowMapper を作成することと同じくらい面倒です。Spring Batch は、JavaBean 仕様を使用して、フィールド名をオブジェクトの setter と照合することにより、フィールドを自動的にマップする FieldSetMapper を提供することにより、これを容易にします。
再びサッカーの例を使用すると、BeanWrapperFieldSetMapper 構成は XML の次のスニペットのようになります。
<bean id="fieldSetMapper"
class="org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper">
<property name="prototypeBeanName" value="player" />
</bean>
<bean id="player"
class="org.springframework.batch.sample.domain.Player"
scope="prototype" /> 再びサッカーの例を使用すると、BeanWrapperFieldSetMapper 構成は Java の次のスニペットのようになります。
@Bean
public FieldSetMapper fieldSetMapper() {
BeanWrapperFieldSetMapper fieldSetMapper = new BeanWrapperFieldSetMapper();
fieldSetMapper.setPrototypeBeanName("player");
return fieldSetMapper;
}
@Bean
@Scope("prototype")
public Player player() {
return new Player();
}FieldSet の各エントリについて、マッパーは、Spring コンテナーがプロパティ名に一致する setter を検索するのと同じ方法で、Player オブジェクトの新しいインスタンスで対応する setter を検索します（このため、プロトタイプスコープが必要です）。FieldSet で使用可能な各フィールドがマップされ、結果の Player オブジェクトが返されます。コードは不要です。
固定長ファイル形式
これまでのところ、区切りファイルのみが詳細に議論されてきました。ただし、それらはファイル読み取りイメージの半分のみを表します。フラットファイルを使用する多くの組織は、固定長形式を使用しています。固定長ファイルの例は次のとおりです。
UK21341EAH4121131.11customer1 UK21341EAH4221232.11customer2 UK21341EAH4321333.11customer3 UK21341EAH4421434.11customer4 UK21341EAH4521535.11customer5
これは 1 つの大きなフィールドのように見えますが、実際には 4 つの異なるフィールドを表します。
ISIN: 並べ替えるアイテムの一意の識別子 -12 文字。
量: 並べ替えるアイテムの数 -3 文字。
価格: 項目の価格 -5 文字の長さ。
お客様: アイテムを並べ替える顧客の ID-9 文字の長さ。
FixedLengthLineTokenizer を構成するときは、これらの各長さを範囲の形式で指定する必要があります。
次の例は、XML で FixedLengthLineTokenizer の範囲を定義する方法を示しています。
<bean id="fixedLengthLineTokenizer"
class="org.springframework.batch.item.file.transform.FixedLengthTokenizer">
<property name="names" value="ISIN,Quantity,Price,Customer" />
<property name="columns" value="1-12, 13-15, 16-20, 21-29" />
</bean>FixedLengthLineTokenizer は前に説明したのと同じ LineTokenizer インターフェースを使用するため、区切り文字が使用されたかのように同じ FieldSet を返します。これにより、BeanWrapperFieldSetMapper の使用など、出力の処理に同じアプローチを使用できます。
範囲の前述の構文をサポートするには、専用のプロパティエディター |
次の例は、Java で FixedLengthLineTokenizer の範囲を定義する方法を示しています。
@Bean
public FixedLengthTokenizer fixedLengthTokenizer() {
FixedLengthTokenizer tokenizer = new FixedLengthTokenizer();
tokenizer.setNames("ISIN", "Quantity", "Price", "Customer");
tokenizer.setColumns(new Range(1, 12),
new Range(13, 15),
new Range(16, 20),
new Range(21, 29));
return tokenizer;
}FixedLengthLineTokenizer は上で説明したのと同じ LineTokenizer インターフェースを使用するため、区切り文字が使用された場合と同じ FieldSet を返します。これにより、BeanWrapperFieldSetMapper を使用するなど、出力の処理に同じアプローチを使用できます。
単一ファイル内の複数のレコード型
ここまでのすべてのファイル読み取りの例は、単純化のためにすべて重要な前提を立てています。ファイル内のすべてのレコードは同じ形式を持っています。ただし、これは常にそうであるとは限りません。ファイルには、異なる形式のレコードがあり、異なるトークン化と異なるオブジェクトへのマッピングが必要になることがよくあります。ファイルからの次の抜粋はこれを示しています。
USER;Smith;Peter;;T;20014539;F LINEA;1044391041ABC037.49G201XX1383.12H LINEB;2134776319DEF422.99M005LI
このファイルには、3 つの型のレコード、"USER"、"LINEA"、"LINEB" があります。"USER" 行は、User オブジェクトに対応します。"LINEA" と "LINEB" は両方とも Line オブジェクトに対応していますが、"LINEA" は "LINEB" よりも多くの情報を持っています。
ItemReader は各行を個別に読み取りますが、ItemWriter が正しい項目を受け取るように、異なる LineTokenizer オブジェクトと FieldSetMapper オブジェクトを指定する必要があります。PatternMatchingCompositeLineMapper は、LineTokenizers へのパターンのマップと FieldSetMappers へのパターンのマップを構成できるようにすることで、これを容易にします。
次の例は、XML で FixedLengthLineTokenizer の範囲を定義する方法を示しています。
<bean id="orderFileLineMapper"
class="org.spr...PatternMatchingCompositeLineMapper">
<property name="tokenizers">
<map>
<entry key="USER*" value-ref="userTokenizer" />
<entry key="LINEA*" value-ref="lineATokenizer" />
<entry key="LINEB*" value-ref="lineBTokenizer" />
</map>
</property>
<property name="fieldSetMappers">
<map>
<entry key="USER*" value-ref="userFieldSetMapper" />
<entry key="LINE*" value-ref="lineFieldSetMapper" />
</map>
</property>
</bean>@Bean
public PatternMatchingCompositeLineMapper orderFileLineMapper() {
PatternMatchingCompositeLineMapper lineMapper =
new PatternMatchingCompositeLineMapper();
Map<String, LineTokenizer> tokenizers = new HashMap<>(3);
tokenizers.put("USER*", userTokenizer());
tokenizers.put("LINEA*", lineATokenizer());
tokenizers.put("LINEB*", lineBTokenizer());
lineMapper.setTokenizers(tokenizers);
Map<String, FieldSetMapper> mappers = new HashMap<>(2);
mappers.put("USER*", userFieldSetMapper());
mappers.put("LINE*", lineFieldSetMapper());
lineMapper.setFieldSetMappers(mappers);
return lineMapper;
} この例では、"LINEA" と "LINEB" には別々の LineTokenizer インスタンスがありますが、両方とも同じ FieldSetMapper を使用します。
PatternMatchingCompositeLineMapper は、各行に正しいデリゲートを選択するために PatternMatcher#match メソッドを使用します。PatternMatcher では、特別な意味を持つ 2 つのワイルドカード文字を使用できます。疑問符（" ？ "）は正確に 1 文字に一致し、アスタリスク（"*" ）は 0 個以上の文字に一致します。上記の構成では、すべてのパターンがアスタリスクで終わり、効果的に行のプレフィックスになることに注意してください。PatternMatcher は、構成の順序に関係なく、常に可能な限り最も具体的なパターンに一致します。"LINE *" と "LINEA *" の両方がパターンとしてリストされている場合、"LINEA" はパターン "LINEA *" に一致し、"LINEB" はパターン "LINE *" に一致します。さらに、単一のアスタリスク（"*" ）は、他のパターンと一致しない行を一致させることにより、デフォルトとして機能できます。
次の例は、XML の他のパターンと一致しない行を一致させる方法を示しています。
<entry key="*" value-ref="defaultLineTokenizer" />次の例は、Java の他のパターンと一致しない行を一致させる方法を示しています。
...
tokenizers.put("*", defaultLineTokenizer());
... トークン化だけに使用できる PatternMatchingCompositeLineTokenizer もあります。
フラットファイルには、それぞれが複数行にわたるレコードが含まれることも一般的です。この状況に対処するには、より複雑な戦略が必要です。この一般的なパターンのデモは、multiLineRecords サンプルにあります。
フラットファイルでの例外処理
行をトークン化すると、例外がスローされる可能性がある多くのシナリオがあります。多くのフラットファイルは不完全であり、誤った形式のレコードが含まれています。多くのユーザーは、課題、元の行、行番号を記録するときに、これらの誤った行をスキップすることを選択します。これらのログは、後で手動で、または別のバッチジョブでインスペクションできます。このため、Spring Batch は、解析例外を処理するための例外の階層 FlatFileParseException および FlatFileFormatException を提供します。FlatFileParseException は、ファイルの読み取り中にエラーが発生した場合に FlatFileItemReader によってスローされます。FlatFileFormatException は、LineTokenizer インターフェースの実装によってスローされ、トークン化中に発生したより具体的なエラーを示します。
IncorrectTokenCountException
DelimitedLineTokenizer と FixedLengthLineTokenizer の両方には、FieldSet の作成に使用できる列名を指定する機能があります。ただし、列名の数が行のトークン化中に見つかった列の数と一致しない場合、FieldSet を作成できず、IncorrectTokenCountException がスローされます。これには、検出されたトークンの数と予想される数が含まれます。次の例:
tokenizer.setNames(new String[] {"A", "B", "C", "D"});
try {
tokenizer.tokenize("a,b,c");
}
catch (IncorrectTokenCountException e) {
assertEquals(4, e.getExpectedCount());
assertEquals(3, e.getActualCount());
} トークナイザーは 4 つの列名で構成されていたが、ファイルで 3 つのトークンしか見つからなかったため、IncorrectTokenCountException がスローされました。
IncorrectLineLengthException
固定長形式でフォーマットされたファイルは、区切り形式とは異なり、各列が事前に定義された幅に厳密に従う必要があるため、解析時に追加の要件があります。行の合計の長さがこの列の最も広い値と等しくない場合、次の例に示すように、例外がスローされます。
tokenizer.setColumns(new Range[] { new Range(1, 5),
new Range(6, 10),
new Range(11, 15) });
try {
tokenizer.tokenize("12345");
fail("Expected IncorrectLineLengthException");
}
catch (IncorrectLineLengthException ex) {
assertEquals(15, ex.getExpectedLength());
assertEquals(5, ex.getActualLength());
} 上記のトークナイザーの構成範囲は、1-5, 6-10,, 11-15 です。その結果、行の合計の長さは 15 です。ただし、前の例では、長さ 5 の行が渡され、IncorrectLineLengthException がスローされました。最初の列のみをマッピングするのではなく、ここで例外をスローすると、FieldSetMapper の列 2 を読み取ろうとして失敗した場合に含まれるよりも多くの情報で、行の処理が早く失敗します。ただし、線の長さが常に一定ではないシナリオがあります。このため、次の例に示すように、'strict' プロパティを使用して行の長さの検証をオフにできます。
tokenizer.setColumns(new Range[] { new Range(1, 5), new Range(6, 10) });
tokenizer.setStrict(false);
FieldSet tokens = tokenizer.tokenize("12345");
assertEquals("12345", tokens.readString(0));
assertEquals("", tokens.readString(1)); 前の例は、tokenizer.setStrict(false) が呼び出されたことを除いて、前の例とほとんど同じです。この設定は、行をトークン化するときに行の長さを強制しないようにトークナイザーに指示します。FieldSet が正しく作成され、返されるようになりました。ただし、残りの値には空のトークンのみが含まれます。
FlatFileItemWriter
フラットファイルへの書き込みには、ファイルからの読み取りで克服しなければならない課題と課題があります。ステップは、トランザクション形式で区切られた形式または固定長形式のいずれかを書き込むことができる必要があります。
LineAggregator
LineTokenizer インターフェースがアイテムを取得して String に変換するために必要であるように、ファイル書き込みでは、ファイルに書き込むために複数のフィールドを単一のストリングに集約する方法が必要です。Spring Batch では、これは LineAggregator であり、次のインターフェース定義に示されています。
public interface LineAggregator<T> {
public String aggregate(T item);
}LineAggregator は LineTokenizer の論理的な反対です。LineTokenizer は String を受け取って FieldSet を返しますが、LineAggregator は item を受け取って String を返します。
PassThroughLineAggregator
LineAggregator インターフェースの最も基本的な実装は PassThroughLineAggregator です。これは、次のコードに示すように、オブジェクトがすでにストリングであるか、そのストリング表現が書き込みに受け入れられることを前提としています。
public class PassThroughLineAggregator<T> implements LineAggregator<T> {
public String aggregate(T item) {
return item.toString();
}
} 上記の実装は、文字列の作成を直接制御する必要があるが、トランザクションや再起動のサポートなどの FlatFileItemWriter の利点が必要な場合に役立ちます。
簡易ファイル作成の例
LineAggregator インターフェースとその最も基本的な実装である PassThroughLineAggregator が定義されたため、基本的な記述の流れを説明できます。
書き込まれるオブジェクトは、
Stringを取得するためにLineAggregatorに渡されます。返された
Stringは、構成されたファイルに書き込まれます。
FlatFileItemWriter からの次の抜粋は、これをコードで表現しています。
public void write(T item) throws Exception {
write(lineAggregator.aggregate(item) + LINE_SEPARATOR);
}XML では、構成の簡単な例は次のようになります。
<bean id="itemWriter" class="org.spr...FlatFileItemWriter">
<property name="resource" value="file:target/test-outputs/output.txt" />
<property name="lineAggregator">
<bean class="org.spr...PassThroughLineAggregator"/>
</property>
</bean>Java では、構成の簡単な例は次のようになります。
@Bean
public FlatFileItemWriter itemWriter() {
return new FlatFileItemWriterBuilder<Foo>()
.name("itemWriter")
.resource(new FileSystemResource("target/test-outputs/output.txt"))
.lineAggregator(new PassThroughLineAggregator<>())
.build();
}FieldExtractor
前述の例は、ファイルへの書き込みの最も基本的な使用に役立つ場合があります。ただし、FlatFileItemWriter のほとんどのユーザーは、書き出す必要があるドメインオブジェクトを持っているため、行に変換する必要があります。ファイルの読み取りでは、以下が必要でした。
ファイルから 1 行読み取ります。
FieldSetを取得するために、その行をLineTokenizer#tokenize()メソッドに渡します。トークン化から返された
FieldSetをFieldSetMapperに渡し、ItemReader#read()メソッドから結果を返します。
ファイルの書き込みには、似ているが逆の手順があります。
書き込むアイテムをライターに渡します。
アイテムのフィールドを配列に変換します。
結果の配列を 1 行に集約します。
フレームワークがオブジェクトのどのフィールドを書き出す必要があるかを知る方法がないため、次のインターフェース定義に示すように、アイテムを配列に変換するタスクを達成するために FieldExtractor を書く必要があります。
public interface FieldExtractor<T> {
Object[] extract(T item);
}FieldExtractor インターフェースの実装では、提供されたオブジェクトのフィールドから配列を作成する必要があります。この配列は、エレメント間の区切り文字を使用して、または固定幅の行の一部として書き出すことができます。
PassThroughFieldExtractor
配列、Collection、FieldSet などのコレクションを書き出す必要がある多くの場合があります。これらのコレクション型の 1 つから配列を「抽出」するのは非常に簡単です。そのためには、コレクションを配列に変換します。このシナリオでは PassThroughFieldExtractor を使用する必要があります。渡されたオブジェクトがコレクションの型ではない場合、PassThroughFieldExtractor は抽出されるアイテムのみを含む配列を返すことに注意してください。
BeanWrapperFieldExtractor
ファイル読み取りのセクションで説明した BeanWrapperFieldSetMapper の場合と同様、変換を自分で記述するよりも、ドメインオブジェクトをオブジェクト配列に変換する方法を構成する方が望ましい場合がよくあります。次の例に示すように、BeanWrapperFieldExtractor はこの機能を提供します。
BeanWrapperFieldExtractor<Name> extractor = new BeanWrapperFieldExtractor<>();
extractor.setNames(new String[] { "first", "last", "born" });
String first = "Alan";
String last = "Turing";
int born = 1912;
Name n = new Name(first, last, born);
Object[] values = extractor.extract(n);
assertEquals(first, values[0]);
assertEquals(last, values[1]);
assertEquals(born, values[2]); このエクストラクターの実装には、必須のプロパティが 1 つだけあります。それは、マップするフィールドの名前です。BeanWrapperFieldSetMapper が FieldSet のフィールドを提供されたオブジェクトの setter にマップするためにフィールド名を必要とするように、BeanWrapperFieldExtractor はオブジェクト配列を作成するために getter にマップするための名前を必要とします。名前の順序によって配列内のフィールドの順序が決まることに注意してください。
区切りファイル書き込みの例
最も基本的なフラットファイル形式は、すべてのフィールドが区切り文字で区切られている形式です。これは、DelimitedLineAggregator を使用して実現できます。次の例では、顧客アカウントのクレジットを表す単純なドメインオブジェクトを書き出します。
public class CustomerCredit {
private int id;
private String name;
private BigDecimal credit;
//getters and setters removed for clarity
} ドメインオブジェクトが使用されているため、使用する区切り文字とともに、FieldExtractor インターフェースの実装を提供する必要があります。
次の例は、XML で区切り文字を使用して FieldExtractor を使用する方法を示しています。
<bean id="itemWriter" class="org.springframework.batch.item.file.FlatFileItemWriter">
<property name="resource" ref="outputResource" />
<property name="lineAggregator">
<bean class="org.spr...DelimitedLineAggregator">
<property name="delimiter" value=","/>
<property name="fieldExtractor">
<bean class="org.spr...BeanWrapperFieldExtractor">
<property name="names" value="name,credit"/>
</bean>
</property>
</bean>
</property>
</bean> 次の例は、Java で区切り文字を使用して FieldExtractor を使用する方法を示しています。
@Bean
public FlatFileItemWriter<CustomerCredit> itemWriter(Resource outputResource) throws Exception {
BeanWrapperFieldExtractor<CustomerCredit> fieldExtractor = new BeanWrapperFieldExtractor<>();
fieldExtractor.setNames(new String[] {"name", "credit"});
fieldExtractor.afterPropertiesSet();
DelimitedLineAggregator<CustomerCredit> lineAggregator = new DelimitedLineAggregator<>();
lineAggregator.setDelimiter(",");
lineAggregator.setFieldExtractor(fieldExtractor);
return new FlatFileItemWriterBuilder<CustomerCredit>()
.name("customerCreditWriter")
.resource(outputResource)
.lineAggregator(lineAggregator)
.build();
} 前の例では、この章で前述した BeanWrapperFieldExtractor を使用して、CustomerCredit 内の名前フィールドとクレジットフィールドをオブジェクト配列に変換し、各フィールドの間にコンマを付けて書き出します。
次の例に示すように、FlatFileItemWriterBuilder.DelimitedBuilder を使用して BeanWrapperFieldExtractor および DelimitedLineAggregator を自動的に作成することもできます。
@Bean
public FlatFileItemWriter<CustomerCredit> itemWriter(Resource outputResource) throws Exception {
return new FlatFileItemWriterBuilder<CustomerCredit>()
.name("customerCreditWriter")
.resource(outputResource)
.delimited()
.delimiter("|")
.names(new String[] {"name", "credit"})
.build();
}固定幅ファイルの書き込みの例
フラットファイル形式の型は、区切り文字だけではありません。多くの人は、各列に設定された幅を使用してフィールド間を線引きすることを好みます。これは通常、「固定幅」と呼ばれます。Spring Batch は、FormatterLineAggregator を使用したファイル書き込みでこれをサポートします。
上記と同じ CustomerCredit ドメインオブジェクトを使用して、XML で次のように構成できます。
<bean id="itemWriter" class="org.springframework.batch.item.file.FlatFileItemWriter">
<property name="resource" ref="outputResource" />
<property name="lineAggregator">
<bean class="org.spr...FormatterLineAggregator">
<property name="fieldExtractor">
<bean class="org.spr...BeanWrapperFieldExtractor">
<property name="names" value="name,credit" />
</bean>
</property>
<property name="format" value="%-9s%-2.0f" />
</bean>
</property>
</bean> 上記と同じ CustomerCredit ドメインオブジェクトを使用して、Java で次のように構成できます。
@Bean
public FlatFileItemWriter<CustomerCredit> itemWriter(Resource outputResource) throws Exception {
BeanWrapperFieldExtractor<CustomerCredit> fieldExtractor = new BeanWrapperFieldExtractor<>();
fieldExtractor.setNames(new String[] {"name", "credit"});
fieldExtractor.afterPropertiesSet();
FormatterLineAggregator<CustomerCredit> lineAggregator = new FormatterLineAggregator<>();
lineAggregator.setFormat("%-9s%-2.0f");
lineAggregator.setFieldExtractor(fieldExtractor);
return new FlatFileItemWriterBuilder<CustomerCredit>()
.name("customerCreditWriter")
.resource(outputResource)
.lineAggregator(lineAggregator)
.build();
}前の例のほとんどは見覚えがあるはずです。ただし、format プロパティの値は新しいものです。
次の例は、XML の format プロパティを示しています。
<property name="format" value="%-9s%-2.0f" />次の例は、Java の format プロパティを示しています。
...
FormatterLineAggregator<CustomerCredit> lineAggregator = new FormatterLineAggregator<>();
lineAggregator.setFormat("%-9s%-2.0f");
... 基礎となる実装は、Java 5 の一部として追加された同じ Formatter を使用して構築されます。Java Formatter は、C プログラミング言語の printf 機能に基づいています。フォーマッタの設定方法の詳細は、フォーマッター (標準 Javadoc) の Javadoc にあります。
次の例に示すように、FlatFileItemWriterBuilder.FormattedBuilder を使用して BeanWrapperFieldExtractor および FormatterLineAggregator を自動的に作成することもできます。
@Bean
public FlatFileItemWriter<CustomerCredit> itemWriter(Resource outputResource) throws Exception {
return new FlatFileItemWriterBuilder<CustomerCredit>()
.name("customerCreditWriter")
.resource(outputResource)
.formatted()
.format("%-9s%-2.0f")
.names(new String[] {"name", "credit"})
.build();
}ファイル作成の処理
FlatFileItemReader は、ファイルリソースと非常に単純な関連にあります。リーダーが初期化されると、ファイルが存在する場合はファイルを開き、存在しない場合は例外をスローします。ファイルの書き込みはそれほど単純ではありません。一見すると、FlatFileItemWriter にも同様の簡単な契約が存在するように見えます。ファイルがすでに存在する場合は例外をスローし、存在しない場合は作成して書き込みを開始します。ただし、Job を再起動すると、問題が発生する機能があります。通常の再起動シナリオでは、契約は逆になります。ファイルが存在する場合は、最後の既知の正常な位置から書き込みを開始し、存在しない場合は例外をスローします。ただし、このジョブのファイル名が常に同じ場合はどうなるでしょうか？ この場合、再起動しない限り、ファイルが存在する場合は削除することをお勧めします。この可能性のため、FlatFileItemWriter にはプロパティ shouldDeleteIfExists が含まれています。このプロパティを true に設定すると、ライターが開かれたときに同じ名前の既存のファイルが削除されます。
XML アイテムリーダーとライター
Spring Batch は、XML レコードの読み取りとそれらの Java オブジェクトへのマッピング、および Java オブジェクトを XML レコードとして書き込むためのトランザクションインフラストラクチャを提供します。
ストリーミング XML の制約 他の標準 XML 解析 API はバッチ処理要件に適合しないため、StAX API は I/O に使用されます（DOM は入力全体を一度にメモリにロードし、SAX はユーザーがコールバックのみを提供できるようにすることで解析プロセスを制御します）。 |
Spring Batch で XML の入出力がどのように機能するかを考慮する必要があります。まず、ファイルの読み取りや書き込みとは異なるが、Spring Batch XML 処理全体で共通する概念がいくつかあります。XML 処理では、トークン化する必要があるレコードの行（FieldSet インスタンス）の代わりに、次の図に示すように、XML リソースは個々のレコードに対応する「フラグメント」のコレクションであると想定されます。
上記のシナリオでは、"trade" タグは「ルート要素」として定義されています。"<trade>" と "</trade>" の間にあるすべてが 1 つの「フラグメント」と見なされます。Spring Batch は、フラグメントをオブジェクトにバインドするためにオブジェクト /XML マッピング (OXM) を使用します。ただし、Spring Batch は特定の XML バインディングテクノロジに縛られていません。一般的な使用方法は、最も一般的な OXM テクノロジの統一された抽象化を提供する Spring OXM に委譲することです。Spring OXM への依存はオプションであり、必要に応じて Spring Batch 固有のインターフェースを実装することもできます。OXM がサポートするテクノロジとの関連を次の図に示します。
OXM の概要と、XML フラグメントを使用してレコードを表現する方法を使用して、リーダーとライターをさらに詳しく調べることができます。
StaxEventItemReader
StaxEventItemReader 構成は、XML 入力ストリームからのレコードを処理するための一般的なセットアップを提供します。まず、StaxEventItemReader が処理できる次の XML レコードのセットを検討します。
<?xml version="1.0" encoding="UTF-8"?>
<records>
<trade xmlns="https://springframework.org/batch/sample/io/oxm/domain">
<isin>XYZ0001</isin>
<quantity>5</quantity>
<price>11.39</price>
<customer>Customer1</customer>
</trade>
<trade xmlns="https://springframework.org/batch/sample/io/oxm/domain">
<isin>XYZ0002</isin>
<quantity>2</quantity>
<price>72.99</price>
<customer>Customer2c</customer>
</trade>
<trade xmlns="https://springframework.org/batch/sample/io/oxm/domain">
<isin>XYZ0003</isin>
<quantity>9</quantity>
<price>99.99</price>
<customer>Customer3</customer>
</trade>
</records>XML レコードを処理できるようにするには、次のものが必要です。
ルート要素名: マップされるオブジェクトを構成するフラグメントのルート要素の名前。構成例は、トレードの価値でこれを示しています。
リソース: 読み取るファイルを表す Spring リソース。
Unmarshaller: XML フラグメントをオブジェクトにマッピングするために Spring OXM によって提供されるアンマーシャリング機能。
次の例は、trade という名前のルート要素、data/iosample/input/input.xml のリソース、XML で tradeMarshaller と呼ばれるアンマーシャラーと連携する StaxEventItemReader を定義する方法を示しています。
<bean id="itemReader" class="org.springframework.batch.item.xml.StaxEventItemReader">
<property name="fragmentRootElementName" value="trade" />
<property name="resource" value="org/springframework/batch/item/xml/domain/trades.xml" />
<property name="unmarshaller" ref="tradeMarshaller" />
</bean> 次の例は、trade という名前のルート要素、data/iosample/input/input.xml のリソース、Java で tradeMarshaller と呼ばれるアンマーシャラーと連携する StaxEventItemReader を定義する方法を示しています。
@Bean
public StaxEventItemReader itemReader() {
return new StaxEventItemReaderBuilder<Trade>()
.name("itemReader")
.resource(new FileSystemResource("org/springframework/batch/item/xml/domain/trades.xml"))
.addFragmentRootElements("trade")
.unmarshaller(tradeMarshaller())
.build();
} この例では、XStreamMarshaller を使用することを選択したことに注意してください。これは、マップとして渡されたエイリアスを受け入れ、最初のキーと値はフラグメントの名前（つまり、ルート要素）とバインドするオブジェクト型です。次に、FieldSet と同様に、オブジェクト型内のフィールドにマップされる他の要素の名前は、マップ内のキーと値のペアとして記述されます。構成ファイルでは、Spring 構成ユーティリティを使用して必要なエイリアスを記述することができます。
次の例は、エイリアスを XML で記述する方法を示しています。
<bean id="tradeMarshaller"
class="org.springframework.oxm.xstream.XStreamMarshaller">
<property name="aliases">
<util:map id="aliases">
<entry key="trade"
value="org.springframework.batch.sample.domain.trade.Trade" />
<entry key="price" value="java.math.BigDecimal" />
<entry key="isin" value="java.lang.String" />
<entry key="customer" value="java.lang.String" />
<entry key="quantity" value="java.lang.Long" />
</util:map>
</property>
</bean>次の例は、Java でエイリアスを記述する方法を示しています。
@Bean
public XStreamMarshaller tradeMarshaller() {
Map<String, Class> aliases = new HashMap<>();
aliases.put("trade", Trade.class);
aliases.put("price", BigDecimal.class);
aliases.put("isin", String.class);
aliases.put("customer", String.class);
aliases.put("quantity", Long.class);
XStreamMarshaller marshaller = new XStreamMarshaller();
marshaller.setAliases(aliases);
return marshaller;
} 入力時に、リーダーは新しいフラグメントが開始されようとしていることを認識するまで XML リソースを読み取ります。デフォルトでは、リーダーは要素名を照合して、新しいフラグメントが開始されようとしていることを認識します。リーダーは、フラグメントからスタンドアロン XML ドキュメントを作成し、そのドキュメントをデシリアライザー（通常は Spring OXM Unmarshaller のラッパー）に渡して、XML を Java オブジェクトにマッピングします。
要約すると、この手順は、Spring 構成によって提供されるインジェクションを使用する次の Java コードに類似しています。
StaxEventItemReader<Trade> xmlStaxEventItemReader = new StaxEventItemReader<>();
Resource resource = new ByteArrayResource(xmlResource.getBytes());
Map aliases = new HashMap();
aliases.put("trade","org.springframework.batch.sample.domain.trade.Trade");
aliases.put("price","java.math.BigDecimal");
aliases.put("customer","java.lang.String");
aliases.put("isin","java.lang.String");
aliases.put("quantity","java.lang.Long");
XStreamMarshaller unmarshaller = new XStreamMarshaller();
unmarshaller.setAliases(aliases);
xmlStaxEventItemReader.setUnmarshaller(unmarshaller);
xmlStaxEventItemReader.setResource(resource);
xmlStaxEventItemReader.setFragmentRootElementName("trade");
xmlStaxEventItemReader.open(new ExecutionContext());
boolean hasNext = true;
Trade trade = null;
while (hasNext) {
trade = xmlStaxEventItemReader.read();
if (trade == null) {
hasNext = false;
}
else {
System.out.println(trade);
}
}StaxEventItemWriter
出力は入力に対して対称的に機能します。StaxEventItemWriter には、Resource、マーシャラー、rootTagName が必要です。Java オブジェクトはマーシャラー（通常は標準の Spring OXM マーシャラー）に渡されます。マーシャラーは、OXM ツールによってフラグメントごとに生成された StartDocument および EndDocument イベントをフィルター処理するカスタムイベントライターを使用して Resource に書き込みます。
次の XML の例では、MarshallingEventWriterSerializer を使用しています。
<bean id="itemWriter" class="org.springframework.batch.item.xml.StaxEventItemWriter">
<property name="resource" ref="outputResource" />
<property name="marshaller" ref="tradeMarshaller" />
<property name="rootTagName" value="trade" />
<property name="overwriteOutput" value="true" />
</bean> 次の Java の例では、MarshallingEventWriterSerializer を使用しています。
@Bean
public StaxEventItemWriter itemWriter(Resource outputResource) {
return new StaxEventItemWriterBuilder<Trade>()
.name("tradesWriter")
.marshaller(tradeMarshaller())
.resource(outputResource)
.rootTagName("trade")
.overwriteOutput(true)
.build();
} 上記の構成では、3 つの必須プロパティを設定し、既存のファイルを上書きできるかどうかを指定するためにこの章で前述したオプションの overwriteOutput=true 属性を設定します。
次の XML の例では、この章で前述した読み取りの例で使用したものと同じマーシャラーを使用しています。
<bean id="customerCreditMarshaller"
class="org.springframework.oxm.xstream.XStreamMarshaller">
<property name="aliases">
<util:map id="aliases">
<entry key="customer"
value="org.springframework.batch.sample.domain.trade.Trade" />
<entry key="price" value="java.math.BigDecimal" />
<entry key="isin" value="java.lang.String" />
<entry key="customer" value="java.lang.String" />
<entry key="quantity" value="java.lang.Long" />
</util:map>
</property>
</bean>次の Java の例では、この章で前述した読み取りの例で使用したものと同じマーシャラーを使用しています。
@Bean
public XStreamMarshaller customerCreditMarshaller() {
XStreamMarshaller marshaller = new XStreamMarshaller();
Map<String, Class> aliases = new HashMap<>();
aliases.put("trade", Trade.class);
aliases.put("price", BigDecimal.class);
aliases.put("isin", String.class);
aliases.put("customer", String.class);
aliases.put("quantity", Long.class);
marshaller.setAliases(aliases);
return marshaller;
}Java の例をまとめると、次のコードは、説明したすべてのポイントを示しており、必要なプロパティのプログラムによる設定を示しています。
FileSystemResource resource = new FileSystemResource("data/outputFile.xml")
Map aliases = new HashMap();
aliases.put("trade","org.springframework.batch.sample.domain.trade.Trade");
aliases.put("price","java.math.BigDecimal");
aliases.put("customer","java.lang.String");
aliases.put("isin","java.lang.String");
aliases.put("quantity","java.lang.Long");
Marshaller marshaller = new XStreamMarshaller();
marshaller.setAliases(aliases);
StaxEventItemWriter staxItemWriter =
new StaxEventItemWriterBuilder<Trade>()
.name("tradesWriter")
.marshaller(marshaller)
.resource(resource)
.rootTagName("trade")
.overwriteOutput(true)
.build();
staxItemWriter.afterPropertiesSet();
ExecutionContext executionContext = new ExecutionContext();
staxItemWriter.open(executionContext);
Trade trade = new Trade();
trade.setPrice(11.39);
trade.setIsin("XYZ0001");
trade.setQuantity(5L);
trade.setCustomer("Customer1");
staxItemWriter.write(trade);JSON アイテムリーダーとライター
Spring Batch は、JSON リソースの読み取りと書き込みを次の形式でサポートします。
[
{
"isin": "123",
"quantity": 1,
"price": 1.2,
"customer": "foo"
},
{
"isin": "456",
"quantity": 2,
"price": 1.4,
"customer": "bar"
}
]JSON リソースは、個々のアイテムに対応する JSON オブジェクトの配列であると想定されています。Spring Batch は特定の JSON ライブラリに関連付けられていません。
JsonItemReader
JsonItemReader は、JSON 解析およびバインディングを org.springframework.batch.item.json.JsonObjectReader インターフェースの実装に委譲します。このインターフェースは、ストリーミング API を使用して JSON オブジェクトをチャンクで読み取ることで実装することを目的としています。現在、2 つの実装が提供されています。
Jackson [GitHub] (英語) から
org.springframework.batch.item.json.JacksonJsonObjectReaderGson [GitHub] (英語) から
org.springframework.batch.item.json.GsonJsonObjectReader
JSON レコードを処理できるようにするには、次のものが必要です。
Resource: 読み取る JSON ファイルを表す Spring リソース。JsonObjectReader: JSON オブジェクトを解析してアイテムにバインドする JSON オブジェクトリーダー
次の例は、以前の JSON リソース org/springframework/batch/item/json/trades.json および Jackson に基づく JsonObjectReader で動作する JsonItemReader を定義する方法を示しています。
@Bean
public JsonItemReader<Trade> jsonItemReader() {
return new JsonItemReaderBuilder<Trade>()
.jsonObjectReader(new JacksonJsonObjectReader<>(Trade.class))
.resource(new ClassPathResource("trades.json"))
.name("tradeJsonItemReader")
.build();
}JsonFileItemWriter
JsonFileItemWriter は、アイテムのマーシャリングを org.springframework.batch.item.json.JsonObjectMarshaller インターフェースに委譲します。このインターフェースの規約は、オブジェクトを取得して JSON String にマーシャリングすることです。現在、2 つの実装が提供されています。
Jackson [GitHub] (英語) から
org.springframework.batch.item.json.JacksonJsonObjectMarshallerGson [GitHub] (英語) から
org.springframework.batch.item.json.GsonJsonObjectMarshaller
JSON レコードを作成できるようにするには、次のものが必要です。
Resource: 書き込む JSON ファイルを表す SpringResourceJsonObjectMarshaller: オブジェクトを JSON 形式にマーシャリングする JSON オブジェクトマーシャラー
次の例は、JsonFileItemWriter を定義する方法を示しています。
@Bean
public JsonFileItemWriter<Trade> jsonFileItemWriter() {
return new JsonFileItemWriterBuilder<Trade>()
.jsonObjectMarshaller(new JacksonJsonObjectMarshaller<>())
.resource(new ClassPathResource("trades.json"))
.name("tradeJsonFileItemWriter")
.build();
}マルチファイル入力
単一の Step 内で複数のファイルを処理することは一般的な要件です。すべてのファイルのフォーマットが同じであると仮定すると、MultiResourceItemReader は XML およびフラットファイル処理の両方でこの型の入力をサポートします。ディレクトリ内の次のファイルを検討してください。
file-1.txt file-2.txt ignored.txt
file-1.txt と file-2.txt は同じ形式であり、ビジネス上の理由から、一緒に処理する必要があります。MuliResourceItemReader は、ワイルドカードを使用して両方のファイルを読み込むために使用できます。
次の例は、XML でワイルドカードを使用してファイルを読み取る方法を示しています。
<bean id="multiResourceReader" class="org.spr...MultiResourceItemReader">
<property name="resources" value="classpath:data/input/file-*.txt" />
<property name="delegate" ref="flatFileItemReader" />
</bean>次の例は、Java でワイルドカードを使用してファイルを読み取る方法を示しています。
@Bean
public MultiResourceItemReader multiResourceReader() {
return new MultiResourceItemReaderBuilder<Foo>()
.delegate(flatFileItemReader())
.resources(resources())
.build();
} 参照されるデリゲートは、単純な FlatFileItemReader です。上記の構成では、両方のファイルから入力を読み取り、ロールバックと再起動のシナリオを処理します。他の ItemReader と同様に、再入力時に余分な入力（この場合はファイル）を追加すると、潜在的な問題が発生する可能性があることに注意してください。バッチジョブは、正常に完了するまで、個々のディレクトリで動作することをお勧めします。
入力リソースは、MultiResourceItemReader#setComparator(Comparator) を使用して順序付けされ、再始動シナリオでのジョブ実行間でリソースの順序が保持されるようにします。 |
データベース
ほとんどのエンタープライズアプリケーションスタイルと同様に、データベースはバッチの主要ストレージメカニズムです。ただし、システムが動作するデータセットのサイズが大きいため、バッチは他のアプリケーションスタイルとは異なります。SQL ステートメントが 100 万行を返す場合、結果セットはすべての行が読み取られるまで、返されたすべての結果をメモリに保持している可能性があります。Spring Batch は、この問題に対して 2 つの型のソリューションを提供します。
カーソルベースの ItemReader 実装
データベースカーソルを使用することは、リレーショナルデータの「ストリーミング」の問題に対するデータベースのソリューションであるため、ほとんどのバッチ開発者のデフォルトのアプローチです。Java ResultSet クラスは、本質的にはカーソルを操作するためのオブジェクト指向のメカニズムです。ResultSet は、現在のデータ行へのカーソルを維持します。ResultSet で next を呼び出すと、このカーソルが次の行に移動します。Spring Batch カーソルベースの ItemReader 実装は、初期化時にカーソルを開き、read の呼び出しごとにカーソルを 1 行前方に移動して、処理に使用できるマップされたオブジェクトを返します。次に、close メソッドが呼び出され、すべてのリソースが解放されます。Spring コア JdbcTemplate は、コールバックパターンを使用して ResultSet のすべての行を完全にマッピングし、メソッド呼び出し元に制御を戻す前に閉じることにより、この問題を回避します。ただし、バッチでは、ステップが完了するまで待機する必要があります。次のイメージは、カーソルベースの ItemReader がどのように機能するかの一般的な図を示しています。この例では SQL を使用していますが（SQL は広く知られているため）、どの技術でも基本的なアプローチを実装できます。
この例は、基本的なパターンを示しています。ID、NAME、BAR という 3 つの列を持つ "FOO" テーブルがある場合、ID が 1 より大きく 7 より小さいすべての行を選択します。これにより、カーソルの先頭（行 1）が ID2 に配置されます。完全にマップされた Foo オブジェクト。read() を再度呼び出すと、カーソルが次の行、つまり ID が 3 の Foo に移動します。これらの読み取りの結果は各 read の後に書き出され、オブジェクトをガベージコレクションできるようになります（インスタンス変数がオブジェクトへの参照を維持していない場合））。
JdbcCursorItemReader
JdbcCursorItemReader は、カーソルベースの手法の JDBC 実装です。ResultSet で直接動作し、DataSource から取得した接続に対して実行する SQL ステートメントが必要です。例として、次のデータベーススキーマを使用します。
CREATE TABLE CUSTOMER (
ID BIGINT IDENTITY PRIMARY KEY,
NAME VARCHAR(45),
CREDIT FLOAT
); 多くの人は各行にドメインオブジェクトを使用することを好むため、次の例では RowMapper インターフェースの実装を使用して CustomerCredit オブジェクトをマップします。
public class CustomerCreditRowMapper implements RowMapper<CustomerCredit> {
public static final String ID_COLUMN = "id";
public static final String NAME_COLUMN = "name";
public static final String CREDIT_COLUMN = "credit";
public CustomerCredit mapRow(ResultSet rs, int rowNum) throws SQLException {
CustomerCredit customerCredit = new CustomerCredit();
customerCredit.setId(rs.getInt(ID_COLUMN));
customerCredit.setName(rs.getString(NAME_COLUMN));
customerCredit.setCredit(rs.getBigDecimal(CREDIT_COLUMN));
return customerCredit;
}
}JdbcCursorItemReader は JdbcTemplate とキーインターフェースを共有するため、ItemReader と対比するために、JdbcTemplate でこのデータを読み込む方法の例を見ると便利です。この例では、CUSTOMER データベースに 1,000 行があると想定しています。最初の例では JdbcTemplate を使用しています。
//For simplicity sake, assume a dataSource has already been obtained
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
List customerCredits = jdbcTemplate.query("SELECT ID, NAME, CREDIT from CUSTOMER",
new CustomerCreditRowMapper()); 上記のコードスニペットを実行した後、customerCredits リストには 1,000 CustomerCredit オブジェクトが含まれます。クエリメソッドでは、DataSource から接続が取得され、指定された SQL が実行され、ResultSet の各行に対して mapRow メソッドが呼び出されます。これを、次の例に示す JdbcCursorItemReader のアプローチと比較してください。
JdbcCursorItemReader itemReader = new JdbcCursorItemReader();
itemReader.setDataSource(dataSource);
itemReader.setSql("SELECT ID, NAME, CREDIT from CUSTOMER");
itemReader.setRowMapper(new CustomerCreditRowMapper());
int counter = 0;
ExecutionContext executionContext = new ExecutionContext();
itemReader.open(executionContext);
Object customerCredit = new Object();
while(customerCredit != null){
customerCredit = itemReader.read();
counter++;
}
itemReader.close(); 上記のコードスニペットを実行した後、カウンターは 1,000 に等しくなります。上記のコードが返された customerCredit をリストに入れていた場合、結果は JdbcTemplate の例とまったく同じになります。ただし、ItemReader の大きな利点は、アイテムを「ストリーミング」できることです。read メソッドは 1 回呼び出すことができ、アイテムは ItemWriter によって書き出され、次のアイテムは read で取得できます。これにより、アイテムの読み取りと書き込みを「チャンク」で実行し、定期的にコミットすることができます。これは、高性能バッチ処理の本質です。さらに、Spring Batch Step への注入用に簡単に構成できます。
次の例は、XML で ItemReader を Step に挿入する方法を示しています。
<bean id="itemReader" class="org.spr...JdbcCursorItemReader">
<property name="dataSource" ref="dataSource"/>
<property name="sql" value="select ID, NAME, CREDIT from CUSTOMER"/>
<property name="rowMapper">
<bean class="org.springframework.batch.sample.domain.CustomerCreditRowMapper"/>
</property>
</bean> 次の例は、Java で ItemReader を Step に注入する方法を示しています。
@Bean
public JdbcCursorItemReader<CustomerCredit> itemReader() {
return new JdbcCursorItemReaderBuilder<CustomerCredit>()
.dataSource(this.dataSource)
.name("creditReader")
.sql("select ID, NAME, CREDIT from CUSTOMER")
.rowMapper(new CustomerCreditRowMapper())
.build();
}追加プロパティ
Java でカーソルを開くための非常に多くのさまざまなオプションがあるため、次の表で説明するように、設定できる JdbcCursorItemReader には多くのプロパティがあります。
ignoreWarnings | SQLWarnings がログに記録されるか、例外を引き起こすかどうかを決定します。デフォルトは |
fetchSize |
|
maxRows | 基になる |
queryTimeout | ドライバーが |
verifyCursorPosition |
|
saveState | リーダーの状態を |
driverSupportsAbsolute | JDBC ドライバーが |
setUseSharedExtendedConnection | カーソルに使用される接続を他のすべての処理で使用する必要があるかどうかを示します。同じトランザクションを共有します。これが |
HibernateCursorItemReader
通常の Spring ユーザーが ORM ソリューションを使用するかどうかについて重要な決定をするように、JdbcTemplate または HibernateTemplate を使用するかどうかに影響するため、Spring Batch ユーザーにも同じオプションがあります。HibernateCursorItemReader は、カーソル手法の Hibernate 実装です。Hibernate のバッチでの使用はかなり物議を醸しています。これは主に、Hibernate が最初にオンラインアプリケーションスタイルをサポートするために開発されたためです。ただし、バッチ処理に使用できないわけではありません。この課題を解決する最も簡単な方法は、標準セッションではなく StatelessSession を使用することです。これにより、Hibernate が採用しているキャッシュおよびダーティチェックがすべて削除され、バッチシナリオで問題が発生する可能性があります。ステートレスセッションと通常の休止状態セッションの違いの詳細については、特定の休止状態リリースのドキュメントを参照してください。HibernateCursorItemReader では、HQL ステートメントを宣言し、SessionFactory を渡すことができます。SessionFactory は、JdbcCursorItemReader と同じ基本的な方法で読み取るために、呼び出しごとに 1 つのアイテムを返します。次の構成例では、JDBC リーダーと同じ「顧客クレジット」の例を使用しています。
HibernateCursorItemReader itemReader = new HibernateCursorItemReader();
itemReader.setQueryString("from CustomerCredit");
//For simplicity sake, assume sessionFactory already obtained.
itemReader.setSessionFactory(sessionFactory);
itemReader.setUseStatelessSession(true);
int counter = 0;
ExecutionContext executionContext = new ExecutionContext();
itemReader.open(executionContext);
Object customerCredit = new Object();
while(customerCredit != null){
customerCredit = itemReader.read();
counter++;
}
itemReader.close(); この構成された ItemReader は、Customer テーブルに対して Hibernate マッピングファイルが正しく作成されていると仮定して、JdbcCursorItemReader で説明されているのとまったく同じ方法で CustomerCredit オブジェクトを返します。'useStatelessSession' プロパティはデフォルトで true に設定されますが、オン / オフを切り替える機能に注意を喚起するためにここで追加されています。また、基になるカーソルのフェッチサイズは、setFetchSize プロパティで設定できることにも注目してください。JdbcCursorItemReader と同様に、構成は簡単です。
次の例は、Hibernate ItemReader を XML で挿入する方法を示しています。
<bean id="itemReader"
class="org.springframework.batch.item.database.HibernateCursorItemReader">
<property name="sessionFactory" ref="sessionFactory" />
<property name="queryString" value="from CustomerCredit" />
</bean> 次の例は、Java で Hibernate ItemReader を注入する方法を示しています。
@Bean
public HibernateCursorItemReader itemReader(SessionFactory sessionFactory) {
return new HibernateCursorItemReaderBuilder<CustomerCredit>()
.name("creditReader")
.sessionFactory(sessionFactory)
.queryString("from CustomerCredit")
.build();
}StoredProcedureItemReader
場合によっては、ストアドプロシージャを使用してカーソルデータを取得する必要があります。StoredProcedureItemReader は JdbcCursorItemReader と同様に機能しますが、カーソルを取得するクエリを実行する代わりに、カーソルを返すストアドプロシージャを実行します。ストアドプロシージャは、3 つの異なる方法でカーソルを返すことができます。
返される
ResultSet（SQL Server、Sybase、DB2、Derby、MySQL で使用）。出力パラメーターとして返される ref-cursor として（Oracle および PostgreSQL で使用）。
ストアド関数呼び出しの戻り値。
次の XML の設定例では、前の例と同じ「顧客クレジット」の例を使用しています。
<bean id="reader" class="o.s.batch.item.database.StoredProcedureItemReader">
<property name="dataSource" ref="dataSource"/>
<property name="procedureName" value="sp_customer_credit"/>
<property name="rowMapper">
<bean class="org.springframework.batch.sample.domain.CustomerCreditRowMapper"/>
</property>
</bean>次の Java の設定例では、前の例と同じ「顧客クレジット」の例を使用しています。
@Bean
public StoredProcedureItemReader reader(DataSource dataSource) {
StoredProcedureItemReader reader = new StoredProcedureItemReader();
reader.setDataSource(dataSource);
reader.setProcedureName("sp_customer_credit");
reader.setRowMapper(new CustomerCreditRowMapper());
return reader;
} 上記の例では、ストアドプロシージャを使用して、返される結果として ResultSet を提供しています（以前のオプション 1）。
ストアドプロシージャが ref-cursor （オプション 2）を返した場合、返された ref-cursor である out パラメーターの位置を指定する必要があります。
次の例は、XML の ref-cursor である最初のパラメーターを操作する方法を示しています。
<bean id="reader" class="o.s.batch.item.database.StoredProcedureItemReader">
<property name="dataSource" ref="dataSource"/>
<property name="procedureName" value="sp_customer_credit"/>
<property name="refCursorPosition" value="1"/>
<property name="rowMapper">
<bean class="org.springframework.batch.sample.domain.CustomerCreditRowMapper"/>
</property>
</bean>次の例は、Java で最初のパラメーターが ref-cursor である場合の操作方法を示しています。
@Bean
public StoredProcedureItemReader reader(DataSource dataSource) {
StoredProcedureItemReader reader = new StoredProcedureItemReader();
reader.setDataSource(dataSource);
reader.setProcedureName("sp_customer_credit");
reader.setRowMapper(new CustomerCreditRowMapper());
reader.setRefCursorPosition(1);
return reader;
} ストアド関数（オプション 3）からカーソルが返された場合は、プロパティ "function" を true に設定する必要があります。デフォルトは false です。
次の例は、XML で true のプロパティを示しています。
<bean id="reader" class="o.s.batch.item.database.StoredProcedureItemReader">
<property name="dataSource" ref="dataSource"/>
<property name="procedureName" value="sp_customer_credit"/>
<property name="function" value="true"/>
<property name="rowMapper">
<bean class="org.springframework.batch.sample.domain.CustomerCreditRowMapper"/>
</property>
</bean> 次の例は、Java での true のプロパティを示しています。
@Bean
public StoredProcedureItemReader reader(DataSource dataSource) {
StoredProcedureItemReader reader = new StoredProcedureItemReader();
reader.setDataSource(dataSource);
reader.setProcedureName("sp_customer_credit");
reader.setRowMapper(new CustomerCreditRowMapper());
reader.setFunction(true);
return reader;
} これらのすべてのケースで、RowMapper、DataSource、実際のプロシージャ名を定義する必要があります。
ストアドプロシージャまたは関数がパラメーターを受け取る場合は、parameters プロパティを使用して宣言および設定する必要があります。次の例では、Oracle の場合、3 つのパラメーターを宣言します。1 つ目は ref-cursor を返す out パラメーターであり、2 つ目と 3 つ目は型 INTEGER の値をとるパラメーターです。
次の例は、XML でパラメーターを操作する方法を示しています。
<bean id="reader" class="o.s.batch.item.database.StoredProcedureItemReader">
<property name="dataSource" ref="dataSource"/>
<property name="procedureName" value="spring.cursor_func"/>
<property name="parameters">
<list>
<bean class="org.springframework.jdbc.core.SqlOutParameter">
<constructor-arg index="0" value="newid"/>
<constructor-arg index="1">
<util:constant static-field="oracle.jdbc.OracleTypes.CURSOR"/>
</constructor-arg>
</bean>
<bean class="org.springframework.jdbc.core.SqlParameter">
<constructor-arg index="0" value="amount"/>
<constructor-arg index="1">
<util:constant static-field="java.sql.Types.INTEGER"/>
</constructor-arg>
</bean>
<bean class="org.springframework.jdbc.core.SqlParameter">
<constructor-arg index="0" value="custid"/>
<constructor-arg index="1">
<util:constant static-field="java.sql.Types.INTEGER"/>
</constructor-arg>
</bean>
</list>
</property>
<property name="refCursorPosition" value="1"/>
<property name="rowMapper" ref="rowMapper"/>
<property name="preparedStatementSetter" ref="parameterSetter"/>
</bean>次の例は、Java でパラメーターを操作する方法を示しています。
@Bean
public StoredProcedureItemReader reader(DataSource dataSource) {
List<SqlParameter> parameters = new ArrayList<>();
parameters.add(new SqlOutParameter("newId", OracleTypes.CURSOR));
parameters.add(new SqlParameter("amount", Types.INTEGER);
parameters.add(new SqlParameter("custId", Types.INTEGER);
StoredProcedureItemReader reader = new StoredProcedureItemReader();
reader.setDataSource(dataSource);
reader.setProcedureName("spring.cursor_func");
reader.setParameters(parameters);
reader.setRefCursorPosition(1);
reader.setRowMapper(rowMapper());
reader.setPreparedStatementSetter(parameterSetter());
return reader;
} パラメーター宣言に加えて、呼び出しのパラメーター値を設定する PreparedStatementSetter 実装を指定する必要があります。これは、上記の JdbcCursorItemReader と同じように機能します。追加プロパティにリストされている追加プロパティはすべて、StoredProcedureItemReader にも適用されます。
ItemReader 実装のページング
データベースカーソルを使用する代わりに、各クエリが結果の一部をフェッチする複数のクエリを実行します。このパートをページと呼びます。各クエリでは、ページで返される開始行番号と行数を指定する必要があります。
JdbcPagingItemReader
ページング ItemReader の実装の 1 つは JdbcPagingItemReader です。JdbcPagingItemReader には、ページを構成する行を取得するために使用される SQL クエリを提供する PagingQueryProvider が必要です。各データベースにはページングサポートを提供する独自の戦略があるため、サポートされるデータベース型ごとに異なる PagingQueryProvider を使用する必要があります。使用されているデータベースを自動検出し、適切な PagingQueryProvider 実装を決定する SqlPagingQueryProviderFactoryBean もあります。これにより、構成が簡素化され、推奨されるベストプラクティスです。
SqlPagingQueryProviderFactoryBean では、select 節と from 節を指定する必要があります。オプションの where 句を指定することもできます。これらの句と必要な sortKey は、SQL ステートメントの構築に使用されます。
実行間でデータが失われないことを保証するために、sortKey に一意のキー制約を設定することが重要です。 |
リーダーが開かれた後、他の ItemReader と同じ基本的な方法で、read への呼び出しごとに 1 つのアイテムを返します。追加の行が必要な場合、ページングはバックグラウンドで発生します。
次の XML の構成例では、前に示したカーソルベースの ItemReader と同様の「顧客クレジット」の例を使用しています。
<bean id="itemReader" class="org.spr...JdbcPagingItemReader">
<property name="dataSource" ref="dataSource"/>
<property name="queryProvider">
<bean class="org.spr...SqlPagingQueryProviderFactoryBean">
<property name="selectClause" value="select id, name, credit"/>
<property name="fromClause" value="from customer"/>
<property name="whereClause" value="where status=:status"/>
<property name="sortKey" value="id"/>
</bean>
</property>
<property name="parameterValues">
<map>
<entry key="status" value="NEW"/>
</map>
</property>
<property name="pageSize" value="1000"/>
<property name="rowMapper" ref="customerMapper"/>
</bean> 次の Java の設定例では、前に示したカーソルベースの ItemReader と同様の「顧客クレジット」の例を使用しています。
@Bean
public JdbcPagingItemReader itemReader(DataSource dataSource, PagingQueryProvider queryProvider) {
Map<String, Object> parameterValues = new HashMap<>();
parameterValues.put("status", "NEW");
return new JdbcPagingItemReaderBuilder<CustomerCredit>()
.name("creditReader")
.dataSource(dataSource)
.queryProvider(queryProvider)
.parameterValues(parameterValues)
.rowMapper(customerCreditMapper())
.pageSize(1000)
.build();
}
@Bean
public SqlPagingQueryProviderFactoryBean queryProvider() {
SqlPagingQueryProviderFactoryBean provider = new SqlPagingQueryProviderFactoryBean();
provider.setSelectClause("select id, name, credit");
provider.setFromClause("from customer");
provider.setWhereClause("where status=:status");
provider.setSortKey("id");
return provider;
} この構成された ItemReader は、指定する必要がある RowMapper を使用して CustomerCredit オブジェクトを返します。'pageSize' プロパティは、クエリの実行ごとにデータベースから読み取られるエンティティの数を決定します。
'parameterValues' プロパティは、クエリのパラメーター値の Map を指定するために使用できます。where 句で名前付きパラメーターを使用する場合、各エントリのキーは、名前付きパラメーターの名前と一致する必要があります。従来の '?' プレースホルダを使用する場合、各エントリのキーは、1 から始まるプレースホルダの番号である必要があります。
JpaPagingItemReader
ページング ItemReader の別の実装は JpaPagingItemReader です。JPA には Hibernate StatelessSession に似た概念がないため、JPA 仕様で提供される他の機能を使用する必要があります。JPA はページングをサポートしているため、バッチ処理に JPA を使用する場合、これは当然の選択です。各ページが読み取られると、エンティティが切り離され、永続コンテキストがクリアされて、ページが処理されるとエンティティをガベージコレクションできるようになります。
JpaPagingItemReader を使用すると、JPQL ステートメントを宣言し、EntityManagerFactory を渡すことができます。次に、呼び出しごとに 1 つのアイテムを返し、他の ItemReader と同じ基本的な方法で読み取ります。追加のエンティティが必要な場合、ページングはバックグラウンドで行われます。
次の XML の設定例では、前に示した JDBC リーダーと同じ "customercredit" の例を使用しています。
<bean id="itemReader" class="org.spr...JpaPagingItemReader">
<property name="entityManagerFactory" ref="entityManagerFactory"/>
<property name="queryString" value="select c from CustomerCredit c"/>
<property name="pageSize" value="1000"/>
</bean>次の Java の設定例では、前に示した JDBC リーダーと同じ "customercredit" の例を使用しています。
@Bean
public JpaPagingItemReader itemReader() {
return new JpaPagingItemReaderBuilder<CustomerCredit>()
.name("creditReader")
.entityManagerFactory(entityManagerFactory())
.queryString("select c from CustomerCredit c")
.pageSize(1000)
.build();
} この構成された ItemReader は、CustomerCredit オブジェクトに正しい JPA アノテーションまたは ORM マッピングファイルがあると仮定して、上記の JdbcPagingItemReader で説明したのとまったく同じ方法で CustomerCredit オブジェクトを返します。'pageSize' プロパティは、クエリ実行ごとにデータベースから読み取られるエンティティの数を決定します。
データベース ItemWriter
フラットファイルと XML ファイルの両方に特定の ItemWriter インスタンスがありますが、データベースの世界にはまったく同じものはありません。これは、トランザクションが必要な機能をすべて提供するためです。ItemWriter の実装は、トランザクションのように動作し、書かれたアイテムを追跡し、適切なタイミングでフラッシュまたはクリアする必要があるため、ファイルに必要です。書き込みはすでにトランザクションに含まれているため、データベースにはこの機能は必要ありません。ユーザーは、ItemWriter インターフェースを実装する独自の DAO を作成するか、一般的な処理の課題のために作成されたカスタム ItemWriter の DAO を使用できます。いずれにしても、課題なく機能するはずです。注意すべきことの 1 つは、出力のバッチ処理によって提供されるパフォーマンスとエラー処理機能です。これは、ItemWriter として休止状態を使用する場合に最も一般的ですが、JDBC バッチモードを使用する場合にも同じ問題が発生する可能性があります。データベース出力のバッチ処理に固有の欠陥はありません。フラッシュに注意し、データにエラーがないことを前提としています。ただし、次の図に示すように、個々のアイテムが例外を引き起こしたかどうか、個々のアイテムが原因であるかどうかを知る方法がないため、書き込み中のエラーは混乱を招く可能性があります。
項目が書き込まれる前にバッファリングされる場合、コミットの直前にバッファがフラッシュされるまでエラーはスローされません。例: 20 個のアイテムがチャンクごとに書き込まれ、15 番目のアイテムが DataIntegrityViolationException をスローすると仮定します。Step に関する限り、エラーが実際に書き込まれるまでエラーが発生したことを知る方法がないため、20 個のアイテムはすべて正常に書き込まれます。Session#flush() が呼び出されると、バッファーが空になり、例外がヒットします。こでは、Step でできることは何もありません。トランザクションをロールバックする必要があります。通常、この例外によりアイテムがスキップされる可能性があり（スキップ / 再試行ポリシーに応じて）、再度書き込まれることはありません。ただし、バッチシナリオでは、どのアイテムが課題を引き起こしたかを知る方法はありません。障害が発生したときにバッファ全体が書き込まれていました。この課題を解決する唯一の方法は、次の図に示すように、各アイテムの後にフラッシュすることです。
これは、特に Hibernate を使用する場合の一般的な使用例であり、ItemWriter の実装の簡単なガイドラインは、write() への各呼び出しでフラッシュすることです。そうすることで、Spring Batch がエラー後の ItemWriter への呼び出しの粒度を内部的に管理して、アイテムを確実にスキップすることができます。
既存のサービスの再利用
バッチシステムは、他のアプリケーションスタイルと組み合わせて使用されることがよくあります。最も一般的なのはオンラインシステムですが、各アプリケーションスタイルで使用する必要なバルクデータを移動することで、統合やシッククライアントアプリケーションをサポートすることもできます。このため、多くのユーザーがバッチジョブ内で既存の DAO またはその他のサービスを再利用したいと考えるのが一般的です。Spring コンテナー自体は、必要なクラスを注入できるようにすることで、これをかなり簡単にします。ただし、別の Spring Batch クラスの依存関係を満たすため、またはそれがステップのメイン ItemReader であるために、既存のサービスが ItemReader または ItemWriter として機能する必要がある場合があります。ラップが必要なサービスごとにアダプタークラスを作成するのは非常に簡単ですが、これは非常に一般的な関心事であるため、Spring Batch は ItemReaderAdapter と ItemWriterAdapter の実装を提供します。どちらのクラスも、デリゲートパターンを呼び出すことにより、標準の Spring メソッドを実装しており、セットアップは非常に簡単です。
次の XML の例では、ItemReaderAdapter を使用しています。
<bean id="itemReader" class="org.springframework.batch.item.adapter.ItemReaderAdapter">
<property name="targetObject" ref="fooService" />
<property name="targetMethod" value="generateFoo" />
</bean>
<bean id="fooService" class="org.springframework.batch.item.sample.FooService" /> 次の Java の例では、ItemReaderAdapter を使用しています。
@Bean
public ItemReaderAdapter itemReader() {
ItemReaderAdapter reader = new ItemReaderAdapter();
reader.setTargetObject(fooService());
reader.setTargetMethod("generateFoo");
return reader;
}
@Bean
public FooService fooService() {
return new FooService();
} 注意すべき重要な点の 1 つは、targetMethod の契約は read の契約と同じでなければならないということです。使い果たされると、null が返されます。それ以外の場合は、Object を返します。それ以外の場合は、ItemWriter の実装に応じて、フレームワークが処理をいつ終了するかを認識できず、無限ループまたは誤った失敗が発生します。
次の XML の例では、ItemWriterAdapter を使用しています。
<bean id="itemWriter" class="org.springframework.batch.item.adapter.ItemWriterAdapter">
<property name="targetObject" ref="fooService" />
<property name="targetMethod" value="processFoo" />
</bean>
<bean id="fooService" class="org.springframework.batch.item.sample.FooService" /> 次の Java の例では、ItemWriterAdapter を使用しています。
@Bean
public ItemWriterAdapter itemWriter() {
ItemWriterAdapter writer = new ItemWriterAdapter();
writer.setTargetObject(fooService());
writer.setTargetMethod("processFoo");
return writer;
}
@Bean
public FooService fooService() {
return new FooService();
}状態の永続性の防止
デフォルトでは、すべての ItemReader および ItemWriter 実装は、コミットされる前に現在の状態を ExecutionContext に保存します。ただし、これが常に望ましい動作であるとは限りません。例: 多くの開発者は、プロセスインジケータを使用してデータベースリーダーを「再実行可能」にすることを選択します。入力データに、処理されたかどうかを示す追加の列が追加されます。特定のレコードが読み取られる (または書き込まれる) と、処理済みフラグが false から true に反転します。その後、SQL ステートメントに where 句に追加のステートメント ( where PROCESSED_IND = false など) を含めることができ、これにより、再起動の場合は未処理のレコードのみが返されるようになります。このシナリオでは、現在の行番号などの状態は再起動時には無関係であるため、保存しないことをお勧めします。このため、すべてのリーダーとライターには 'saveState' プロパティが含まれています。
次の Bean 定義は、XML での状態の永続化を防ぐ方法を示しています。
<bean id="playerSummarizationSource" class="org.spr...JdbcCursorItemReader">
<property name="dataSource" ref="dataSource" />
<property name="rowMapper">
<bean class="org.springframework.batch.sample.PlayerSummaryMapper" />
</property>
<property name="saveState" value="false" />
<property name="sql">
<value>
SELECT games.player_id, games.year_no, SUM(COMPLETES),
SUM(ATTEMPTS), SUM(PASSING_YARDS), SUM(PASSING_TD),
SUM(INTERCEPTIONS), SUM(RUSHES), SUM(RUSH_YARDS),
SUM(RECEPTIONS), SUM(RECEPTIONS_YARDS), SUM(TOTAL_TD)
from games, players where players.player_id =
games.player_id group by games.player_id, games.year_no
</value>
</property>
</bean>次の Bean 定義は、Java で状態の永続性を防ぐ方法を示しています。
@Bean
public JdbcCursorItemReader playerSummarizationSource(DataSource dataSource) {
return new JdbcCursorItemReaderBuilder<PlayerSummary>()
.dataSource(dataSource)
.rowMapper(new PlayerSummaryMapper())
.saveState(false)
.sql("SELECT games.player_id, games.year_no, SUM(COMPLETES),"
+ "SUM(ATTEMPTS), SUM(PASSING_YARDS), SUM(PASSING_TD),"
+ "SUM(INTERCEPTIONS), SUM(RUSHES), SUM(RUSH_YARDS),"
+ "SUM(RECEPTIONS), SUM(RECEPTIONS_YARDS), SUM(TOTAL_TD)"
+ "from games, players where players.player_id ="
+ "games.player_id group by games.player_id, games.year_no")
.build();
} 上記で設定された ItemReader は、参加する実行に対して ExecutionContext にエントリを作成しません。
カスタム ItemReader および ItemWriter の作成
これまでのところ、この章では、Spring Batch での読み取りと書き込みの基本的な契約と、そのための一般的な実装について説明しました。ただし、これらはすべてかなり一般的であり、すぐに使用可能な実装ではカバーされない可能性のある多くの潜在的なシナリオがあります。このセクションでは、簡単な例を使用して、カスタム ItemReader および ItemWriter 実装を作成し、それらの契約を正しく実装する方法を示します。ItemReader は、リーダーまたはライターを再起動可能にする方法を示すために、ItemStream も実装しています。
カスタム ItemReader の例
この例の目的のために、提供されたリストから読み取る単純な ItemReader 実装を作成します。次のコードに示すように、ItemReader の最も基本的な契約である read メソッドを実装することから始めます。
public class CustomItemReader<T> implements ItemReader<T> {
List<T> items;
public CustomItemReader(List<T> items) {
this.items = items;
}
public T read() throws Exception, UnexpectedInputException,
NonTransientResourceException, ParseException {
if (!items.isEmpty()) {
return items.remove(0);
}
return null;
}
} 上記のクラスは項目のリストを取得し、一度に 1 つずつ返し、リストからそれぞれを削除します。リストが空の場合、次のテストコードに示すように、null を返し、ItemReader の最も基本的な要件を満たします。
List<String> items = new ArrayList<>();
items.add("1");
items.add("2");
items.add("3");
ItemReader itemReader = new CustomItemReader<>(items);
assertEquals("1", itemReader.read());
assertEquals("2", itemReader.read());
assertEquals("3", itemReader.read());
assertNull(itemReader.read());ItemReader を再起動可能にする
最後の課題は、ItemReader を再起動可能にすることです。現在、処理が中断されて再び開始される場合、ItemReader は最初から開始する必要があります。これは実際には多くのシナリオで有効ですが、バッチジョブを中断したところから再開することが望ましい場合があります。多くの場合、重要な判別基準は、リーダーがステートフルかステートレスかです。ステートレスリーダーは再起動の可能性について心配する必要はありませんが、ステートフルリーダーは再起動時に最後の既知の状態を再構成する必要があります。このため、可能な場合はカスタムリーダーをステートレスにしておくことをお勧めします。そのため、再起動性について心配する必要はありません。
状態を保存する必要がある場合は、ItemStream インターフェースを使用する必要があります。
public class CustomItemReader<T> implements ItemReader<T>, ItemStream {
List<T> items;
int currentIndex = 0;
private static final String CURRENT_INDEX = "current.index";
public CustomItemReader(List<T> items) {
this.items = items;
}
public T read() throws Exception, UnexpectedInputException,
ParseException, NonTransientResourceException {
if (currentIndex < items.size()) {
return items.get(currentIndex++);
}
return null;
}
public void open(ExecutionContext executionContext) throws ItemStreamException {
if (executionContext.containsKey(CURRENT_INDEX)) {
currentIndex = new Long(executionContext.getLong(CURRENT_INDEX)).intValue();
}
else {
currentIndex = 0;
}
}
public void update(ExecutionContext executionContext) throws ItemStreamException {
executionContext.putLong(CURRENT_INDEX, new Long(currentIndex).longValue());
}
public void close() throws ItemStreamException {}
}ItemStream update メソッドを呼び出すたびに、ItemReader の現在のインデックスが、提供された ExecutionContext に "current.index" のキーで格納されます。ItemStream open メソッドが呼び出されると、ExecutionContext がチェックされ、そのキーを持つエントリが含まれているかどうかが確認されます。キーが見つかった場合、現在のインデックスはその場所に移動されます。これはかなり些細な例ですが、それでも一般的な契約を満たしています。
ExecutionContext executionContext = new ExecutionContext();
((ItemStream)itemReader).open(executionContext);
assertEquals("1", itemReader.read());
((ItemStream)itemReader).update(executionContext);
List<String> items = new ArrayList<>();
items.add("1");
items.add("2");
items.add("3");
itemReader = new CustomItemReader<>(items);
((ItemStream)itemReader).open(executionContext);
assertEquals("2", itemReader.read()); ほとんどの ItemReader には、はるかに洗練された再起動ロジックがあります。たとえば、JdbcCursorItemReader は、カーソル内で最後に処理された行の行 ID を格納します。
また、ExecutionContext 内で使用されるキーは些細なものであってはならないことに注意する価値があります。これは、同じ ExecutionContext が Step 内のすべての ItemStreams に使用されるためです。ほとんどの場合、一意性を保証するには、単にクラス名をキーに追加するだけで十分です。ただし、まれに同じステップで同じ型の ItemStream が 2 つ使用される場合（出力に 2 つのファイルが必要な場合に発生する可能性があります）、より一意の名前が必要です。このため、Spring Batch ItemReader および ItemWriter 実装の多くには、このキー名をオーバーライドできる setName() プロパティがあります。
カスタム ItemWriter の例
カスタム ItemWriter の実装は、上記の ItemReader の例と多くの点で似ていますが、独自の例を保証するのに十分な点で異なります。ただし、再起動可能性の追加は基本的に同じであるため、この例では説明しません。ItemReader の例と同様に、できるだけ簡単に例を保つために List が使用されます。
public class CustomItemWriter<T> implements ItemWriter<T> {
List<T> output = TransactionAwareProxyFactory.createTransactionalList();
public void write(List<? extends T> items) throws Exception {
output.addAll(items);
}
public List<T> getOutput() {
return output;
}
}ItemWriter を再起動可能にする
ItemWriter を再起動可能にするには、ItemReader と同じプロセスに従い、実行コンテキストを同期する ItemStream インターフェースを追加および実装します。この例では、処理されたアイテムの数をカウントし、それをフッターレコードとして追加する必要があります。それを行う必要がある場合は、ItemWriter に ItemStream を実装して、ストリームが再び開かれた場合にカウンターが実行コンテキストから再構成されるようにすることができます。
多くの現実的なケースでは、カスタム ItemWriter は、それ自体が再起動可能な別のライター（たとえば、ファイルへの書き込み時）に委譲するか、トランザクションリソースに書き込むため、ステートレスなので再起動する必要はありません。ステートフルライターを使用している場合は、ItemWriter と同様に ItemStream も実装する必要があります。また、ライターのクライアントは ItemStream を認識する必要があるため、構成でストリームとして登録する必要があることも覚えておいてください。
アイテムリーダーおよびライターの実装
このセクションでは、前のセクションでまだ説明していないリーダーとライターを紹介します。
デコレーター
場合によっては、ユーザーが既存の ItemReader に追加される特別な動作が必要です。Spring Batch は、ItemReader および ItemWriter の実装に追加の動作を追加できる、すぐに使えるデコレーターを提供します。
Spring Batch には次のデコレーターが含まれます。
SynchronizedItemStreamReader
スレッドセーフではない ItemReader を使用する場合、Spring Batch は SynchronizedItemStreamReader デコレータを提供します。これは、ItemReader スレッドセーフにするために使用できます。Spring Batch は、SynchronizedItemStreamReaderBuilder を提供して、SynchronizedItemStreamReader のインスタンスを構築します。
SingleItemPeekableItemReader
Spring Batch には、ItemReader にピークメソッドを追加するデコレーターが含まれています。このピークメソッドにより、ユーザーは 1 つ先のアイテムをピークできます。ピークを繰り返し呼び出すと同じアイテムが返され、これが read メソッドから返される次のアイテムです。Spring Batch は、SingleItemPeekableItemReader のインスタンスを構築するために SingleItemPeekableItemReaderBuilder を提供します。
| SingleItemPeekableItemReader のピークメソッドは、複数のスレッドでピークを尊重することができないため、スレッドセーフではありません。ピークしたスレッドの 1 つだけが、次の読み取り呼び出しでそのアイテムを取得します。 |
SynchronizedItemStreamWriter
スレッドセーフではない ItemWriter を使用する場合、Spring Batch は SynchronizedItemStreamWriter デコレータを提供します。これは、ItemWriter スレッドセーフにするために使用できます。Spring Batch は、SynchronizedItemStreamWriterBuilder を提供して、SynchronizedItemStreamWriter のインスタンスを構築します。
MultiResourceItemWriter
MultiResourceItemWriter は ResourceAwareItemWriterItemStream をラップし、現在のリソースに書き込まれたアイテムの数が itemCountLimitPerResource を超えると、新しい出力リソースを作成します。Spring Batch は、MultiResourceItemWriterBuilder を提供して、MultiResourceItemWriter のインスタンスを構築します。
メッセージングリーダーおよびライター
Spring Batch は、一般的に使用されるメッセージングシステム用に次のリーダーとライターを提供します。
AmqpItemReader
AmqpItemReader は、AmqpTemplate を使用して交換機からメッセージを受信または変換する ItemReader です。Spring Batch は、AmqpItemReaderBuilder を提供して、AmqpItemReader のインスタンスを構築します。
AmqpItemWriter
AmqpItemWriter は、AmqpTemplate を使用して AMQP 交換にメッセージを送信する ItemWriter です。指定された AmqpTemplate で名前が指定されていない場合、メッセージは名前のない交換に送信されます。Spring Batch は、AmqpItemWriter のインスタンスを構築するために AmqpItemWriterBuilder を提供します。
JmsItemReader
JmsItemReader は、JmsTemplate を使用する JMS 用の ItemReader です。テンプレートには、read() メソッドにアイテムを提供するために使用されるデフォルトの宛先が必要です。Spring Batch は、JmsItemReader のインスタンスを構築するために JmsItemReaderBuilder を提供します。
JmsItemWriter
JmsItemWriter は、JmsTemplate を使用する JMS 用の ItemWriter です。テンプレートには、write(List) でアイテムを送信するために使用されるデフォルトの宛先が必要です。Spring Batch は、JmsItemWriter のインスタンスを構築するために JmsItemWriterBuilder を提供します。
データベースリーダー
Spring Batch は、次のデータベースリーダーを提供します。
Neo4jItemReader
Neo4jItemReader は、ページング技術を使用してグラフデータベース Neo4j からオブジェクトを読み取る ItemReader です。Spring Batch は、Neo4jItemReader のインスタンスを構築する Neo4jItemReaderBuilder を提供します。
MongoItemReader
MongoItemReader は、ページング技術を使用して MongoDB からドキュメントを読み取る ItemReader です。Spring Batch は、MongoItemReader のインスタンスを構築する MongoItemReaderBuilder を提供します。
HibernateCursorItemReader
HibernateCursorItemReader は、Hibernate 上に構築されたデータベースレコードを読み取るための ItemStreamReader です。HQL クエリを実行し、初期化されると、read() メソッドが呼び出されると結果セットを反復処理し、現在の行に対応するオブジェクトを連続して返します。Spring Batch は、HibernateCursorItemReaderBuilder を提供して、HibernateCursorItemReader のインスタンスを構築します。
データベースライター
Spring Batch は、次のデータベースライターを提供します。
Neo4jItemWriter
Neo4jItemWriter は、Neo4j データベースに書き込む ItemWriter 実装です。Spring Batch は、Neo4jItemWriter のインスタンスを構築する Neo4jItemWriterBuilder を提供します。
MongoItemWriter
MongoItemWriter は、Spring Data の MongoOperations の実装を使用して MongoDB ストアに書き込む ItemWriter 実装です。Spring Batch は、MongoItemWriterBuilder を提供して、MongoItemWriter のインスタンスを構築します。
RepositoryItemWriter
RepositoryItemWriter は、Spring Data からの CrudRepository の ItemWriter ラッパーです。Spring Batch は、RepositoryItemWriterBuilder を提供して、RepositoryItemWriter のインスタンスを構築します。
HibernateItemWriter
HibernateItemWriter は、Hibernate セッションを使用して、現在の Hibernate セッションの一部ではないエンティティを保存または更新する ItemWriter です。Spring Batch は、HibernateItemWriter のインスタンスを構築する HibernateItemWriterBuilder を提供します。
JdbcBatchItemWriter
JdbcBatchItemWriter は、NamedParameterJdbcTemplate のバッチ機能を使用して、提供されたすべてのアイテムに対してステートメントのバッチを実行する ItemWriter です。Spring Batch は、JdbcBatchItemWriterBuilder を提供して、JdbcBatchItemWriter のインスタンスを構築します。
専門リーダー
Spring Batch は、次の専門リーダーを提供します。
LdifReader
LdifReader は、Resource から LDIF（LDAP Data Interchange Format）レコードを読み取り、解析し、実行された read ごとに LdapAttribute オブジェクトを返します。Spring Batch は、LdifReader のインスタンスを構築するために LdifReaderBuilder を提供します。
専門ライター
Spring Batch は、以下の専門ライターを提供しています。