データフローDSL考 (3)

ニコニコ超会議 2012の超エンジニアミーティングのScalaユーザーグループのコマで「Scalaでプログラムを作りました」のセッションを行いましたが、右の図はそのスライドで用いたDSLの分類です。

ここでは、DSLを分類する軸として、用途の軸と実現方法の軸を用いています。

用途の軸の項目は以下の2つです。

• モデル
• フレームワークAPI

実現方法の軸の項目は以下の2つです。

• 内部DSL
• 外部DSL

一口にDSLといっても2×2で4種類のDSLに分類することができるわけです。各象限には、セッションで取り上げた技術を分類して配置しています。

Scalaは、関数型言語の機能と文法上の工夫で内部DSLの実現に適した言語となっています。またパーサーコンビネータを用いることで外部DSLの実現も簡単に行うことができます。

Scalaの名前の由来である「Scalable Language」は、言語をドメイン向けに拡張していくことができる、ということを意味していると思いますが、そういう意味で、まさにDSL指向言語ということができるでしょう。

データフローDSLという観点で、今までScalaで2つの内部DSLをつくってきました。一つはモデル記述の私家版Asakusa DSL、もう一つはg3向けのフレームワークAPIです。

まず、データフローDSLのモデル記述の実現例である私家版Asakusa DSLについてみていきます。

私家版Asakusa DSL

Asakusa Framework(以下Asakusa)は、基幹業務システムのバッチを高速処理するためのHadoopフレームワークです。単にHadoopを使いやすくしているだけでなく、ジョブ管理などを含めた基幹業務に必要なミドルウェアの基盤を提供している点が特徴です。

Hadoopのベースになった技術はMapReduceですが、mapとreduceという用語からも分かるとおり、関数型的なデータフロー演算が、計算モデルのベースになっています。オブジェクト指向と関数型の接点はデータフローになりそうだというお話をしてきましたが、クラウド・プラットフォームと基幹業務の接点もデータフローということになりそうです。

オブジェクト指向、関数型といった開発手法の方向からも、プラットフォームや業務ドメインからの方向からも、データフローに収斂する構図になっており、データフローは一種のホットスポットとなっています。

そういう意味でも、データフローDSLの果たす役割は今後大きくなりそうです。その一つの応用として、Asakusa向けのScala DSLは格好の例題となります。

Scala DSL

私家版Asakusa DSL(以下Scala DSL)は、現在Javaベースで提供されているAsakusa DSLのScala版の試作品です。去年の3月頃に作成しました。(「Asakusa Scala DSL」)

Scala DSLで記述したモデルは以下のようになります。

view plainprint?

1. package sample  
2.   
3. import org.simplemodeling.dsl.domain._  
4. import org.simplemodeling.dsl.flow._  
5.   
6. class 仕入明細データ extends DomainResource  
7. class 仕入返品データ extends DomainResource  
8. class 費用振替データ extends DomainResource  
9. class 売価変更データ extends DomainResource  
10.   
11. class 修正在庫振替TRN extends DataSet  
12. class 修正未収収益TRN extends DataSet  
13. class 修正在庫移動TRN extends DataSet  
14. class 未払計上TRN extends DataSet  
15.   
16. class 仕入TRN extends DataSet  
17. class 在庫振替TRN extends DataSet  
18. class 在庫移動TRN extends DataSet  
19. class 未収収益TRN extends DataSet  
20.   
21. class 計上済仕入TRN extends DataSet  
22. class 計上済未収収益TRN extends DataSet  
23. class 計上済未払費用TRN extends DataSet  
24. class 更新済買掛残高TRN extends DataSet  
25.   
26. class 請求エラーTRN extends DataSet  
27. class 支払不可消込TRN extends DataSet  
28. class 支払可消込TRN extends DataSet  
29. class 照合済支払費用TRN extends DataSet  
30. class 照合済未収収益TRN extends DataSet  
31. class 照合済仕入TRN extends DataSet  
32. class 照合済請求TRN extends DataSet  
33.   
34. class 仕入データ extends DataSource4[仕入明細データ, 仕入返品データ,  
35.                                      費用振替データ, 売価変更データ]  
36. class 修正データ extends DataSource4[修正在庫振替TRN, 修正未収収益TRN,  
37.                                      修正在庫移動TRN, 未払計上TRN]  
38. class 売価変更在庫変更TRN extends DataSet  
39. class 仕入データTRN extends DataSource4[仕入TRN, 在庫振替TRN,  
40.                                         在庫移動TRN, 未収収益TRN]  
41. class 残高更新TRN extends DataSource4[計上済仕入TRN, 計上済未収収益TRN,  
42.                                       計上済未払費用TRN, 更新済買掛残高TRN]  
43. class 請求TRN extends DataSet  
44. class 会計データTRN extends DataSource7[請求エラーTRN, 支払不可消込TRN, 支払可消込TRN,  
45.                                         照合済支払費用TRN, 照合済未収収益TRN,  
46.                                         照合済仕入TRN, 照合済請求TRN]  
47.   
48. case class 仕入データ取り込み(cout: Port[売価変更在庫変更TRN]) extends Operator12[仕入データ, 仕入データTRN, 売価変更在庫変更TRN](cout)  
49. case class 残高更新(cin: Port[修正データ]) extends Operator21[仕入データTRN, 修正データ, 残高更新TRN](cin)  
50. case class 照合処理(cin: Port[請求TRN]) extends Operator21[残高更新TRN, 請求TRN, 会計データTRN](cin)  
51.   
52. // サブフロー用に追加したオペレーション  
53. case object 売価変更在庫変更TRN修正 extends Operator11[売価変更在庫変更TRN, 売価変更在庫変更TRN]  
54.   
55. // サブフロー用に追加したオペレーション  
56. case object 修正データ追加修正 extends Operator11[修正データ, 修正データ]  
57.   
58. // 図7 改善された会計処理バッチの処理フロー  
59. case class 会計処理バッチ extends Flow32[仕入データ, 修正データ, 請求TRN,  
60.                                     会計データTRN, 売価変更在庫変更TRN] {  
61. // 追加したサブフロー1  
62. // 処理結果を会計処理バッチの2番目の出力ポートに出力                                        
63.   val 売価変更在庫変更TRN補正 = new Flow11[売価変更在庫変更TRN, 売価変更在庫変更TRN] {  
64.     start op11(売価変更在庫変更TRN修正) end(会計処理バッチ.this.out2)  
65.   }  
66.   
67. // 追加したサブフロー2  
68. // 会計処理バッチの2番目の入力ポートからデータを入力  
69.   val 修正データ補正 = new Flow11[修正データ, 修正データ] {  
70.     start(会計処理バッチ.this.in2) op11(修正データ追加修正) end  
71.   }  
72.   
73. // 仕入データ取り込みの2番目の出力先をサブフロー1に変更  
74. // 残高更新の2番目の入力元をサブフロー2に変更  
75.   start op12(仕入データ取り込み(売価変更在庫変更TRN補正.post1)) op21(残高更新(修正データ補正.get1)) op21(照合処理(in3)) end  
76. }  

package sample

import org.simplemodeling.dsl.domain._
import org.simplemodeling.dsl.flow._

class 仕入明細データ extends DomainResource
class 仕入返品データ extends DomainResource
class 費用振替データ extends DomainResource
class 売価変更データ extends DomainResource

class 修正在庫振替TRN extends DataSet
class 修正未収収益TRN extends DataSet
class 修正在庫移動TRN extends DataSet
class 未払計上TRN extends DataSet

class 仕入TRN extends DataSet
class 在庫振替TRN extends DataSet
class 在庫移動TRN extends DataSet
class 未収収益TRN extends DataSet

class 計上済仕入TRN extends DataSet
class 計上済未収収益TRN extends DataSet
class 計上済未払費用TRN extends DataSet
class 更新済買掛残高TRN extends DataSet

class 請求エラーTRN extends DataSet
class 支払不可消込TRN extends DataSet
class 支払可消込TRN extends DataSet
class 照合済支払費用TRN extends DataSet
class 照合済未収収益TRN extends DataSet
class 照合済仕入TRN extends DataSet
class 照合済請求TRN extends DataSet

class 仕入データ extends DataSource4[仕入明細データ, 仕入返品データ,
                                     費用振替データ, 売価変更データ]
class 修正データ extends DataSource4[修正在庫振替TRN, 修正未収収益TRN,
                                     修正在庫移動TRN, 未払計上TRN]
class 売価変更在庫変更TRN extends DataSet
class 仕入データTRN extends DataSource4[仕入TRN, 在庫振替TRN,
                                        在庫移動TRN, 未収収益TRN]
class 残高更新TRN extends DataSource4[計上済仕入TRN, 計上済未収収益TRN,
                                      計上済未払費用TRN, 更新済買掛残高TRN]
class 請求TRN extends DataSet
class 会計データTRN extends DataSource7[請求エラーTRN, 支払不可消込TRN, 支払可消込TRN,
                                        照合済支払費用TRN, 照合済未収収益TRN,
                                        照合済仕入TRN, 照合済請求TRN]

case class 仕入データ取り込み(cout: Port[売価変更在庫変更TRN]) extends Operator12[仕入データ, 仕入データTRN, 売価変更在庫変更TRN](cout)
case class 残高更新(cin: Port[修正データ]) extends Operator21[仕入データTRN, 修正データ, 残高更新TRN](cin)
case class 照合処理(cin: Port[請求TRN]) extends Operator21[残高更新TRN, 請求TRN, 会計データTRN](cin)

// サブフロー用に追加したオペレーション
case object 売価変更在庫変更TRN修正 extends Operator11[売価変更在庫変更TRN, 売価変更在庫変更TRN]

// サブフロー用に追加したオペレーション
case object 修正データ追加修正 extends Operator11[修正データ, 修正データ]

// 図7 改善された会計処理バッチの処理フロー
case class 会計処理バッチ extends Flow32[仕入データ, 修正データ, 請求TRN,
                                    会計データTRN, 売価変更在庫変更TRN] {
// 追加したサブフロー1
// 処理結果を会計処理バッチの2番目の出力ポートに出力                                      
  val 売価変更在庫変更TRN補正 = new Flow11[売価変更在庫変更TRN, 売価変更在庫変更TRN] {
    start op11(売価変更在庫変更TRN修正) end(会計処理バッチ.this.out2)
  }

// 追加したサブフロー2
// 会計処理バッチの2番目の入力ポートからデータを入力
  val 修正データ補正 = new Flow11[修正データ, 修正データ] {
    start(会計処理バッチ.this.in2) op11(修正データ追加修正) end
  }

// 仕入データ取り込みの2番目の出力先をサブフロー1に変更
// 残高更新の2番目の入力元をサブフロー2に変更
  start op12(仕入データ取り込み(売価変更在庫変更TRN補正.post1)) op21(残高更新(修正データ補正.get1)) op21(照合処理(in3)) end
}

Scala DSLは(仮に)SimpleModelerに組み込んで、データフロー図を表示できるようにしています。このモデルからSimpleModelerで生成したデータフロー図は以下のものになります。

このデータフローDSLは、静的型付けによってデータの入出力関係やパラメタの個数の間違いをエラーチェックできることが特徴です。このこともあり、入出力のデータの定義を詳細に行なっています。

しかし、メインとなるデータフローの記述は以下の一行に収まっています。op12は入力が1つ、出力が2つあるプロセス、op21は入力が2つ、出力が1つのプロセスです。大枠は「仕入データ取り込み」を行った後「残高更新」する処理となります。

「仕入データ取り込み」はメインの出力に加えて「売価変更在庫変更TRN補正」サブフローに出力を行います。また、「残高更新」はメインの入力に加えて「修正データ補正」サブフローからの入力を行います。

view plainprint?

1. start op12(仕入データ取り込み(売価変更在庫変更TRN補正.post1)) op21(残高更新(修正データ補正.get1)) op21(照合処理(in3)) end  

start op12(仕入データ取り込み(売価変更在庫変更TRN補正.post1)) op21(残高更新(修正データ補正.get1)) op21(照合処理(in3)) end

データフローを構成するサブフローは以下の「売価変更在庫変更TRN補正」と「修正データ補正」の2つです。それぞれのサブフローも一行に収まっています。

view plainprint?

1. val 売価変更在庫変更TRN補正 = new Flow11[売価変更在庫変更TRN, 売価変更在庫変更TRN] {  
2.     start op11(売価変更在庫変更TRN修正) end(会計処理バッチ.this.out2)  
3.   }  

val 売価変更在庫変更TRN補正 = new Flow11[売価変更在庫変更TRN, 売価変更在庫変更TRN] {
    start op11(売価変更在庫変更TRN修正) end(会計処理バッチ.this.out2)
  }

view plainprint?

1. val 修正データ補正 = new Flow11[修正データ, 修正データ] {  
2.     start(会計処理バッチ.this.in2) op11(修正データ追加修正) end  
3.   }  

val 修正データ補正 = new Flow11[修正データ, 修正データ] {
    start(会計処理バッチ.this.in2) op11(修正データ追加修正) end
  }

評価

Scalaの内部DSLでデータフローDSLを設計する上での論点としては以下のものが挙げられます。

• パイプラインの記法を取り入れるか否か(ノード記述方式を採用するか否か)
• 静的型付けをどこまで使うか
• 2つのフローの接続方式
• ロジック記述方式

私家版Asakusa DSLは、実験的な意味もあり、パイプライン記法と静的型付けの方に大きく倒した方式にしてみました。そこそこ、うまくいっているのではないかというのが自己評価で、パイプライン記法と静的型付けの組合せは一応実用化の目処がついたと判断しています。

「2つのフローの接続方法」は、サブフローを直接指定する方式にしてみましたが、今の目で見ると、チャネル方式などを導入してサブフロー間を疎結合にしていくアプローチの方がよさそうです。

「ロジック記述方式」は、今回の試作では入れませんでしたが、非常に重要な項目です。内部DSLを採用するメリットの一つがロジックをホスト言語で直接記述できる点にあります。

実用化

実用化に際しては、サブフローを集めた部品を、他の部品と合成してさらに大きくしていく開発手法が求められるようになると思います。サブフロー部品の汎用部品化も視野に入ってきそうです。サブフローの接続方式はチャネル方式の方がこういった開発手法に適していそうなので、改良候補となります。

静的型付け&ジェネリック型&関数型は、こういった部品の合成時のメカニズムおよび合成時のエラーチェックにも威力を発揮しそうです。そういう意味でも、直接ロジック記述することができるメリットも含めて、Scalaをホスト言語にした内部DSL方式は、非常に有力な方式ですね。

SimpleModeler

私家版Asakusa DSLは、仮にSimpleModelerに組み込んでみましたが、Asakusaに限らず汎用的なデータフローDSLとして利用できそうな感触を持っています。

懸案事項はサブフローの接続方式とロジック記述方式です。

サブフローの接続方式はチャネル方式にするとして、問題はロジック記述方式です。これは、型クラスを用いて実現できそうな気がしているのですが、このあたりが次のチャレンジとなります。

折を見てこれらの改良を施した上で、Asakusaはもちろん、他のプラットフォーム向けのコード生成にもトライしてみたいと考えています。