mwSoft blog

HadoopのContextは、「nextKeyValue」「getCurrentKey」「getCurrentValue」の3つを持っていてループで回して使ったりする。

この子をScalaのforでこんな風に回してみたかった。

for ((key, value) <- context) {
  // ここで処理
}

ScalaはhasNextとnextさえ実装出来ればIteratorとして扱えるらしい。けど、ContextにはhasNext的なものがない。nextKeyValueがBooleanを返す。

こういう場合にどうやって書くのがいいのか分からなかったので参考にIteratorのソースを見てたら、singleという1個だけ要素を取るらしきメソッドを見かける。

def single[A](elem: A) = new Iterator[A] {
  private var hasnext = true
  def hasNext: Boolean = hasnext
  def next(): A =
    if (hasnext) { hasnext = false; elem }
    else empty.next()
}

これと同じ感じでいいか、ということでこんなんで済ませてみた。思うところはあれど。ていうかこれちゃんといろんなケースで動くんだろうか。

implicit def hadoopContext2scalaIterator[A, B](that: MapContext[A, B, _, _]) = new Iterator[(A, B)] {
  var callnext = true
  var callhasnext = false
  var hasnext = true
  def hasNext: Boolean = {
    callhasnext = true
    if (callnext) { callnext = false; that.nextKeyValue() }
    else hasnext
  }

  def next(): (A, B) = {
    if (!callhasnext) hasnext = that.nextKeyValue()
    callnext = true
    callhasnext = false
    if (hasnext) (that.getCurrentKey(), that.getCurrentValue())
    else Iterator.empty.next
  }
}

これでMapperのrunをオーバーライドすれば、WordCountがこんな感じで書けたり。

override def run(context: Context) {
  for ((key, value) <- context; word <- value.toString.split("\\s"))
    context.write(new Text(word), new LongWritable(1L))
}

setupとかcleanup呼び出すよりrunを継承した方が書きやすい時ってあるよね。