package com.odaff.samza.yarn

  

import org.apache.hadoop.yarn.conf.YarnConfiguration
import org.apache.samza.config.Config
import org.apache.samza.job.StreamJobFactory
import org.apache.samza.job.yarn.YarnJob
import org.apache.samza.util.hadoop.HttpFileSystem
import scala.collection.JavaConversions._

class RemoteYarnJobFactory extends StreamJobFactory {
  def getJob(config: Config) = {
    val hConfig = new YarnConfiguration

    config.filterKeys(_.startsWith("yarn.")).foreach(kv => hConfig.set(kv._1, kv._1))

    hConfig.set("fs.http.impl", classOf[HttpFileSystem].getName)
    hConfig.set("fs.https.impl", classOf[HttpFileSystem].getName)

    new YarnJob(config, hConfig)
  }
}
