AfwHS α版

Asakusa FrameworkをScalaで記述するライブラリー「AfwHS」α版。

概要

AfwHSは、Scalaで記述したコードをAsakusaFWの（Java）DSLに変換して実行できるようにするライブラリー。

AfwHSはコードネーム「AsakusaFramework Hishidama Scala」の略かなぁ(苦笑)
「Hishidama」なんて恥ずかしい単語が入っている理由は、AsakusaFWでScalaと言えばasami224さんの「Asakusa Scala DSL」の事を指すので、違いを出す必要があるから…。
「AfwHS」を何て発音すればいいか分からないから(爆)、「AHS」でいいかも。

DMDLで作ったJavaのモデルクラスを元に、AsakusaFWの演算子（アノテーション）を模したScalaの関数（これがAfwHSで用意されているもの）を使って処理を記述する。
それを元にAsakusaFWのOperator・JobFlowを生成する。

とりあえず作ってみたって感じなので、α版。

WordCountの例

AfwHSでWordCountを書くと、こんな感じ。

import scala.collection.JavaConverters._

import jp.hishidama.afwhs._

import sample.wordcount.modelgen.dmdl.model.LineModel
import sample.wordcount.modelgen.dmdl.model.LineModelWrapper
import sample.wordcount.modelgen.dmdl.model.LineModelImplicit._
import sample.wordcount.modelgen.dmdl.model.WordCountModel
import sample.wordcount.modelgen.dmdl.model.WordCountModelWrapper
import sample.wordcount.modelgen.dmdl.model.WordCountModelImplicit._

object WordCount {

  def run(in: Source[LineModel]) = {
    val split = in.cogroup("split", "", Result[WordCountModel]("out")) { (s, r) =>
      s.asScala.foreach {
        _.getTextAsString().split("[ \t]+").foreach { word =>
          r.add(WordCountModelWrapper(word, 1))
        }
      }
    }
    val count = split.cogroup("sum", "word", Result[WordCountModel]("out")) { (s, r) =>
      val word  = s.get(0).getWordAsString()
      val count = s.asScala.map(_.getCount).sum
      r.add(WordCountModelWrapper(word, count))
    }

    count //戻り値
  }

青い字はDMDLによって生成されたモデルクラスやそのメソッドそのもの。
オレンジ色のはモデルのScala用ラッパークラス。これもモデルドライバーの機能で生成される。

Source/ResultやcogoupはAsakusaFWのクラス（DSL）を模したもので、AfwHSで用意しているクラスやメソッド。

s0はデータモデルのjava.util.Listで、CoGroupの引数と全く同じ。
r0も同じくCoGroupで戻り値を返すResultそのもの。

その他は普通のScalaのメソッド・関数。

このrunメソッドを呼び出す方法が2種類ある。

  def main(args: Array[String]) {
    if (!args.exists(_ == "generate")) {

      // 普通にデータを渡して実行する
      val slist = List("Hello World", "Hello Asakusa", "Hello Hadoop")
      val mlist = slist.map { text => LineModelWrapper(text).asJava }
      val r = run(Source(mlist))
      println(r)

    } else {
      // Asakusa DSLを生成する
      val gen = new Generator("C:/workspace", "afwhs-wordcount", "sample.wordcount", "WordCount")
      val r = run(gen.newSource())
      gen.out(r)

      gen.prepareGenerate()
      gen.generateOperator()
      gen.generateImporter()
      gen.generateExporter()
      gen.generateFlow()
      gen.generateFlowTest()
    }
  }
}

とりあえず、実行時の引数に「generate」という文字列が入っているかどうかで処理を分岐している。

→もうちょっと詳しいWordCountの例

変更履歴