【iDempiere/Pentaho Lab】iDempiereからETLを呼び出す

　iDempiereから、PentahoのETLツールであるKettle(ケトル)のSpoon(スプーン)で作成した、トランスフォーメーションやジョブを、引数を渡して実行する方法を調査＆研究したいと思います。これが、ちゃんと実行できるとリアルタイムで他のアプリケーションとのデータのやりとりもかなり楽に実現できるのではないかと思います。

　iDempiereの受注伝票(Sales Order)のデータを保存する時に、PentahoのPanで実行する事ができるトランスショーメーションのBatファイルを呼び出して、別のテーブル(C_Order2)に受注伝票のヘッダー情報(C_Orderテーブルの情報)を書き込むというシナリオで調査＆研究したいと思います。

　このシナリオの検証ポイントは下記になるかなと思います。

• JavaからトランスフォーメーションをキックするBatファイルの実行。
• Batファイルに引数として、受注伝票ID(C_Order_ID)を渡して、それをさらにETLに渡す。
• 結果を受け取って、失敗した際にはiDempiere側でも失敗にする必要がある。

といったところでしょうか。

環境は、とりあえずローカルのWindoiws7で行いたいと思います。

　Get System Infoステップを使用して、Windowsのコマンドプロンプトに入力した引数を受け取れるように準備しておきます。

　コマンドプロンプトで入力した引数はGet System InfoステップでC_Order_IDというフィールド名で受け取っているので、それを"?"を使用して、Table inputステップに渡します。

　Table input ステップで取得したデータを、Insert/Update ステップを使用してC_Order2テーブルにインサートもしくはアップデートして行きます。

　次のようなバッチファイルを作成します。ここではバッチファイルの名前は"From_Order_To_Order2.bat"として、C:\pentaho-ce\data-integration\wsにおいておきます。

../Pan.bat /file "C:\pentaho-ce\ws\From_Order_to_Order2.ktr"  %1 /level:Basic > C:\pentaho-ce\ktr.log

【バッチファイルの構成】

• ..\Pan.bat ⇒ 実行するバッチファイルの指定。
• /file "C:\pentaho-ce\ws\From_Order_to_Order2.ktr" %1 ⇒ 実行するトランスフォーメーションのファイルの指定と、引数の設定。%1でコマンドプロンプトで入力した引数を取得する。
• /level:Basic > C:\pentaho-ce\ktr.log ⇒ ログファイルの指定。ログ出力は、Erroe / Notihg / Minimal / Basic / Detailed / Debug / Rowlevel が指定できます。 

　この状態でコマンドプロンプトより下記のように、batファイルと引数(1000033)を入力して実行してみます。

　そうすると、次のようなエラーがログファイルに出力されました。

ERROR: operator does not exist: numeric = character varying  ヒント: No operator matches the given name and argument type(s). You might need to add explicit type casts.  ポジション: 1397

どうやら、キャストするように言われているようです。そこで、select Valuesのステップを１つ追加します。

　C_Order_IDのフィールドは、文字列としてコマンドプロンプトよりデータを受け取っているので、それを数値としてキャストします。

　これで、C_Order_ID=1000033のレコードがC_Order2テーブルにインサート（もしくはアップデート）された事が確認できると思います。

　Javaでコーディングする前に、検証を簡単にするために今まで相対パスで指定していた所は絶対パスに修正しておきます。またログファイルのリダイレクト設定は外しておきます。

"C:\pentaho-ce\data-integration\Pan.bat" /file "C:\pentaho-ce\ws\From_Order_to_Order2.ktr" %1 /level:Basic

　Javaから、batファイルを実行するには以下のように記述するようです。

try{

String C_Order_id = new Integer( getC_Order_ID() ).toString();

String[] command = new String[]{"C:\\pentaho-ce\\data-integration\\ws\\From_Order_To_Order2.bat", C_Order_id };

ProcessBuilder pb = new ProcessBuilder(command);

Process process = pb.start();

InputStream is = process.getInputStream();

BufferedReader br  = new BufferedReader(new InputStreamReader(is));

String stringLine = null;

StringBuffer stringBuffer = new StringBuffer();

while ((stringLine  = br.readLine())!=null)

{

System.out.println("pentaho>"+stringLine );

stringBuffer =stringBuffer.append(stringLine).append(System.getProperty("line.separator"));

}

is.close();int ret = process.waitFor();

} catch ( Throwable t ){

}

　これで、いちおうiDempiereからPentahoのETLツールKettleのSpoonで作成したトランスフォーメーションを、呼び出して実行し結果を取得する事ができます。結果を取得できるので、それでハンドリングすれば、エラー処理もできるのではないかと思います。

　ここまで試してみて、この処理には大きな欠陥がある事に気が付きました。iDempiereでは、Compiereとモデルクラスのコミット処理が大きく異なっていたので、受注伝票の処理を行っているMOrderクラスのafterSave()メソッドでデータベースからデータを抽出するのでは、C_OrderテーブルとC_Order2テーブルのデータは同期できないという事です。

　iDempiereでは、afterSave()メソッドの処理も問題なく完了した時点でコミットされるようなので、aftesrSave()メソッドでのデータを抽出してしまうと、C_Order2のテーブルにはC_Orderテーブルの処理中のレコードのコミット前の状態のデータがインサートもしくはアップデートされる事になってしまうのです。

【Compiere Lab】モデルクラス内でのトランザクション処理の実装(2012/6/13)

【iDempiere Lab】iDempiereのモデルクラス内でのトランザクション処理(2013/8/17)

　ここで、Step1に戻って再度トランスフォーメーションから作り直したいと思います。

　今度作成するトランスフォーメーションは、DBからのデータ抽出ステップは排除して、Get System Info ステップで、C_Order2テーブルにインサートする値を引数として渡してもらう処理にしようと思います。

　ただ、このステップでは、コマンドライン引数として渡せる数は10個に限られるのが弱点ですね。無制限に取得できる方法があると良いのですが…。iDempiereの複数の値を区切り文字を付加して１つの引数にまとめて、その引数をETL側で区切り文字をもとに分割して戻すという処理が、簡単に思いつく方法でしょうか…。

　コマンドプロンプトから渡される情報は、文字情報なのでキャストする設定を行います。

そして、そのままデータをインサートもしくはアップデートします。

バッチファイルも引数を複数取得できるように修正します。

"C:\pentaho-ce\data-integration\Pan.bat" /file "C:\pentaho-ce\ws\From_Order_to_Order2.ktr" %1 %2 %3 %4・・・ /level:Basic

　Javaのプログラムも複数引数を渡すように修正します。

String C_Order_ID = new Integer( getC_Order_ID() ).toString();

String AD_Client_ID = new Integer(getAD_Client_ID() ).toString();

String AD_Org_ID = new  Integer(getAD_Org_ID() ).toString();

String C_BPartner_ID =  new  Integer(getC_BPartner_ID() ).toString();

…

String[] command = new String[]{"C:\\pentaho-ce\\data-integration\\ws\\From_Order_To_Order2.bat", C_Order_ID,AD_Client_ID,AD_Org_ID,C_BPartner_ID,…};

　これで、C_OrderテーブルとC_Order2テーブルの内容をリアルタイムで同期させる事ができると思います。

　もし、ETL側で何らかのエラーが発生しデータが正しく登録されなかった場合は、iDempiereはそれを判断し、afterSave()メソッドでretrun falseをすればエラー処理も2つのテーブル間での同期を崩すことなく正しく処理できるのではないかと思います。

　pentaho側でエラーが発生した場合はロールバックして、iDempiereにエラーを返す必要があります。ここでは、pentahoのロールバック処理について追記しておきます。

　その他タブを選択し、"固有の接続を使用する"をONにします。"固有の接続を使用する"というのは適切な訳では無いそうで、このフラグをONにする事により、pentaho側の処理が正常終了でない場合はロールバックがされるそうです。

次の事については今後の課題としたいと思います。

• エラー処理の複雑さの緩和…エラー処理のハンドリングがこのままでは、正規表現などでデータを抜き取って判断する必要があるため複雑になってしまう。もっと簡素化したいなと思います。
• afterSave()メソッドで実装するのではなく、ワークフロー上でプロセスを実行する事で上手く動作させる事ができないかな？->ワークフロー上でプロセスを実行して、エラーをハンドリングして、データの整合性が保てれば、既存ソースコードを修正する事無くできるので理想的。

Pan User Documentation

オープンソースBIのPentaho（ペンタホ）ブログ：よくある質問(PDI編)

オープンソースBIのPentaho（ペンタホ）ブログ：Pentaho Date Integration（PDI） コマンドライン引数を指定して実行する

技術的に自由なブログ：初めてのPentaho Data Integration 5 パラメータ(Variable)

Sakanaya_tarohの日記　javaコードによるLinuxコマンドの実行について(2013/1/29)

前のページに戻る