HBase OutputFormat for Apache Flink

HBaseOutputFormat.scala

import org.apache.flink.api.common.io.OutputFormat
import org.apache.flink.configuration.Configuration
import org.apache.hadoop.hbase._
import client._
import util.Bytes

import language.{ implicitConversions, reflectiveCalls }

import java.math.BigDecimal
import java.nio.ByteBuffer

abstract class HBaseOutputFormat[T](tableNames: String*)
                                   (implicit config: Map[String, String])
  extends OutputFormat[T] {

  var tables = Map.empty[String, HTable]

  def configure(parameters: Configuration) = ()

  def open(taskNumber: Int, numTasks: Int) = {
    val hConf = HBaseConfiguration.create
    for { (key, value) <- config if key.startsWith("hbase") }
      hConf.set(key, value)
    for { name <- tableNames } tables += name -> new HTable(hConf, name)
  }

  def close() = try withResources(tables.values.toSeq: _*) {
    _.foreach { _.flushCommits() }
  } finally tables = Map.empty

  def withResource [R <: { def close(): Unit }](resource:  R )
                                               (f: R      => Unit) =
    try f(resource) finally resource.close()

  def withResources[R <: { def close(): Unit }](resources: R*)
                                               (f: Seq[R] => Unit) =
    resources.foldRight(f) { (r, f) =>
      _ => withResource(r) { _ => f(resources) }
    } (resources)
}