Added ability to add to Spark's hadoop configuration, as well as a path to get and test the configs

Author: Stephen Carman

job-server/src/spark.jobserver/JobManagerActor.scala

@@ -1,17 +1,21 @@
 package spark.jobserver
 
 import java.util.concurrent.Executors._
-import akka.actor.{ActorRef, Props, PoisonPill}
+
+import akka.actor.{ActorRef, PoisonPill, Props}
 import com.typesafe.config.Config
 import java.net.{URI, URL}
 import java.util.concurrent.atomic.AtomicInteger
+
 import ooyala.common.akka.InstrumentedActor
-import org.apache.spark.{ SparkEnv, SparkContext }
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.{SparkConf, SparkContext, SparkEnv}
 import org.joda.time.DateTime
-import scala.concurrent.{ Future, ExecutionContext }
+
+import scala.concurrent.{ExecutionContext, Future}
 import scala.util.{Failure, Success, Try}
 import spark.jobserver.ContextSupervisor.StopContext
-import spark.jobserver.io.{JobDAOActor, JobDAO, JobInfo, JarInfo}
+import spark.jobserver.io.{JarInfo, JobDAO, JobDAOActor, JobInfo}
 import spark.jobserver.util.{ContextURLClassLoader, SparkJobUtils}
 
 object JobManagerActor {
@@ -20,9 +24,11 @@ object JobManagerActor {
   case class StartJob(appName: String, classPath: String, config: Config,
                       subscribedEvents: Set[Class[_]])
   case class KillJob(jobId: String)
+  case object ContextConfig
   case object SparkContextStatus
 
   // Results/Data
+  case class ContextConfig(contextName: String, contextConfig: SparkConf, hadoopConfig: Configuration)
   case class Initialized(contextName: String, resultActor: ActorRef)
   case class InitError(t: Throwable)
   case class JobLoadingError(err: Throwable)
@@ -152,7 +158,23 @@ class JobManagerActor(contextConfig: Config) extends InstrumentedActor {
           sender ! SparkContextAlive
         } catch {
           case e: Exception => {
-            logger.error("SparkContext is not exist!")
+            logger.error("SparkContext does not exist!")
+            sender ! SparkContextDead
+          }
+        }
+      }
+    }
+    case ContextConfig => {
+      if (jobContext.sparkContext == null){
+        sender ! SparkContextDead
+      }else{
+        try {
+          val conf: SparkConf = jobContext.sparkContext.getConf
+          val hadoopConf: Configuration = jobContext.sparkContext.hadoopConfiguration
+          sender ! ContextConfig(jobContext.sparkContext.appName, conf, hadoopConf)
+        } catch {
+          case e: Exception => {
+            logger.error("SparkContext does not exist!")
             sender ! SparkContextDead
           }
         }

job-server/src/spark.jobserver/context/SparkContextFactory.scala

@@ -50,9 +50,11 @@ class DefaultSparkContextFactory extends SparkContextFactory {
   type C = SparkContext with ContextLike
 
   def makeContext(sparkConf: SparkConf, config: Config,  contextName: String): C = {
-    new SparkContext(sparkConf) with ContextLike {
+    val sc = new SparkContext(sparkConf) with ContextLike {
       def sparkContext: SparkContext = this
       def isValidJob(job: SparkJobBase): Boolean = job.isInstanceOf[SparkJob]
     }
+    for ((k, v) <- SparkJobUtils.getHadoopConfig(config)) sc.hadoopConfiguration.set(k, v)
+    sc
   }
 }

job-server/src/spark.jobserver/util/SparkJobUtils.scala

@@ -73,6 +73,17 @@ object SparkJobUtils {
     conf
   }
 
+  /**
+    *
+    * @param config the specific context configuration
+    * @return a map of the hadoop configuration values or an empty Map
+    */
+  def getHadoopConfig(config: Config): Map[String, String] = {
+    Try(config.getConfig("hadoop").entrySet().asScala.map { e =>
+      e.getKey -> e.getValue.unwrapped().toString
+    }.toMap).getOrElse(Map())
+  }
+
   /**
    * Returns the maximum number of jobs that can run at the same time
    */

job-server/test/spark.jobserver/LocalContextSupervisorSpec.scala

@@ -1,10 +1,12 @@
 package spark.jobserver
 
 import akka.actor._
-import akka.testkit.{TestKit, ImplicitSender}
+import akka.testkit.{ImplicitSender, TestKit}
 import com.typesafe.config.ConfigFactory
-import spark.jobserver.io.{JobDAOActor, JobDAO}
-import org.scalatest.{Matchers, FunSpecLike, BeforeAndAfterAll, BeforeAndAfter}
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.SparkConf
+import spark.jobserver.io.{JobDAO, JobDAOActor}
+import org.scalatest.{BeforeAndAfter, BeforeAndAfterAll, FunSpecLike, Matchers}
 
 import scala.concurrent.duration._
 
@@ -36,6 +38,9 @@ object LocalContextSupervisorSpec {
           spark.driver.allowMultipleContexts = true
           spark.ui.enabled = false
         }
+        hadoop {
+          mapreduce.framework.name = "ayylmao"
+        }
       }
     }
     akka.log-dead-letters = 0
@@ -71,6 +76,7 @@ class LocalContextSupervisorSpec extends TestKit(LocalContextSupervisorSpec.syst
   }
 
   import ContextSupervisor._
+  import JobManagerActor._
 
   describe("context management") {
     it("should list empty contexts at startup") {
@@ -99,6 +105,21 @@ class LocalContextSupervisorSpec extends TestKit(LocalContextSupervisorSpec.syst
       rActor.path.toString should not include ("global")
     }
 
+    it("should be able to get context configs") {
+      supervisor ! AddContext("c1", contextConfig)
+      expectMsg(ContextInitialized)
+      supervisor ! GetContext("c1")
+      expectMsgPF(5 seconds, "I can't find that context :'-(") {
+        case (contextActor: ActorRef, resultActor: ActorRef) => {
+          contextActor ! ContextConfig
+          val cc = expectMsgClass(classOf[ContextConfig])
+          cc.contextName shouldBe "c1"
+          cc.contextConfig.get("spark.ui.enabled") shouldBe "false"
+          cc.hadoopConfig.get("mapreduce.framework.name") shouldBe "ayylmao"
+        }
+      }
+    }
+
     it("should be able to stop contexts already running") {
       supervisor ! AddContext("c1", contextConfig)
       expectMsg(ContextInitialized)