［Talend］CSVを読み込んでカラムを追加してCSVを出力するジョブ定義

最初に

Talendにはたくさんのコンポーネントがあって、どれを使っていいのか最初は全然分からず戸惑う方が多いのではないでしょうか。
そこで、よく使いそうなコンポーネント（処理）を不定期ではありますがコツコツと紹介していけたらと思います。
今回は【特定のディレクトリに存在するCSVを読み込んでその行を処理した時間を追加して処理結果を特定のディレクトリに出力する】について順を追って定義していきたいと思います。
サンプルデータは以下のデータを使います。

0348,201505,CA,500000,TEST,BTJ（Dom）
0349,201505,CA,500000,TEST1,BTJ（Pre）
0349,201505,CA,500000,TEST2,BTJ（Pre）
0353,201506,AC,400000,TEST3,KHK
0349,201506,CA,1000000,TEST4,BTJ（Pre）
0349,201506,CA,1000000,TEST5,BTJ（Pre）
0348,201506,CA,1000000,TEST6,BTJ（Dom）
0354,201506,CA,300000,TEST7,FR
0354,201506,CA,400000,TEST8,FR
0354,201506,CA,300000,TEST9,FR
0347,201506,AC,250000,TEST10,DM（LT）

Talendのコンポーネントはどれを使うのか？

まずTalendを実行しているPCのローカルディスクに存在するCSVファイルを読み込むコンポーネントとしてtFileInputDelimitedを使います。
次に読み込んだデータ（行）に対してカラムを追加するので、コンポーネントとしてtMapを使って処理時間のカラムを追加します。
最後の処理として処理時間が追加されたCSVデータをPCのローカルディスクに書き出すコンポーネントとしてtFileOutputDelimitedを使います。
この３つのコンポーネントを利用することで超簡単ではありますがTalendジョブ定義ができあがります。

実際にTalendのジョブ定義をしてみる

tFileInputDelimitedの定義手順

tFileInputDelimitedコンポーネントをドラッグ＆ドロップでステージに置きます。その後読み込み対象とするCSVファイル名を指定します。必要に応じて行区切り記号とフィールド区切り記号を変更します。

次にスキーマの編集を行い、読み込み対象のデータ定義を行います。

上記でtFileInputDelimitedの定義は完了です。

tMapの定義手順

tMapコンポーネントをドラッグ＆ドロップでステージに置きます。次にtFileInputDelimitedコンポーネントからtMapに処理が流れるように右クリックで処理フロー線を定義します。（ここではメインを選択して処理フロー線を引きました。）

処理フロー定義をすると下記のように流れができます。

次にtMapコンポーネントをダブルクリックすると下記のようなマッピングが定義できる画面が出てきます。
画面右ペインにある＋をクリックして出力テーブルの定義を追加します。あとは処理時間の日付を出力テーブルへ追加したいので、CreateDateカラムを更に追加して関数の定義をします。

読み込んだデータの内、Remark,Enable_Month,Target,Buget,PrjNmを出力データとします。（Target2の列データは不要なので削ぎ落とします）
ここまでの作業が完了したら左から右へ対象となるカラムをドラッグ＆ドロップして出力データ行のマッピングを行います。

上記でtMapの定義は完了です。

tFileOutputDelimitedの定義手順

tFileOutputDelimitedコンポーネントをドラッグ＆ドロップでステージに置きます。その後出力対象とするCSVファイル名を指定します。必要に応じて行区切り記号とフィールド区切り記号を変更します。

tFileInputDelimitedコンポーネントの時と同じようにスキーマの編集を行い、終了したらtMapコンポーネントからtFileOutputDelimitedコンポーネントに処理フロー線を定義します。

これで一連のジョブ定義が完了です。

ジョブを実行してみる

実行はどちらかのボタンをクリックすれば可能です。

エラーなく処理が完了すれば下記のような画面が表示されます。

ジョブ実行で作成されたCSVファイルは以下の内容になっているはずです。

0348,201505,CA,500000,BTJ（Dom）,2015-07-06 23:20:33
0349,201505,CA,500000,BTJ（Pre）,2015-07-06 23:20:33
0349,201505,CA,500000,BTJ（Pre）,2015-07-06 23:20:33
0353,201506,AC,400000,KHK,2015-07-06 23:20:33
0349,201506,CA,1000000,BTJ（Pre）,2015-07-06 23:20:33
0349,201506,CA,1000000,BTJ（Pre）,2015-07-06 23:20:33
0348,201506,CA,1000000,BTJ（Dom）,2015-07-06 23:20:33
0354,201506,CA,300000,FR,2015-07-06 23:20:33
0354,201506,CA,400000,FR,2015-07-06 23:20:33
0354,201506,CA,300000,FR,2015-07-06 23:20:33
0347,201506,AC,250000,DM（LT）,2015-07-06 23:20:33

期待通り読み込んだCSVのTarget2が削除されて、最後のカラムに処理時間が追加されていました。

最後に

とてもシンプルなジョブ定義をやってみました。次回は今回のジョブ定義に処理を追加してAWSのS3バケットに書き出すものをやってみたいと思います。