Issue #0000 fix: Cassandra connector fix for ETBCoverageSummarizer.

project-sunbird · Dec 20, 2018 · 3721fa6 · 3721fa6
1 parent 3449139
commit 3721fa6
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 2 deletions.
diff --git a/...mework/analytics-core/src/main/scala/org/ekstep/analytics/framework/util/CommonUtil.scala b/...mework/analytics-core/src/main/scala/org/ekstep/analytics/framework/util/CommonUtil.scala
@@ -43,7 +43,7 @@ object CommonUtil {
         config.parallelization.getOrElse(defParallelization);
     }
 
-    def getSparkContext(parallelization: Int, appName: String): SparkContext = {
+    def getSparkContext(parallelization: Int, appName: String, sparkCassandraConnectionHost: Option[AnyRef] = None): SparkContext = {
         JobLogger.log("Initializing Spark Context")
         val conf = new SparkConf().setAppName(appName);
         val master = conf.getOption("spark.master");
@@ -60,6 +60,11 @@ object CommonUtil {
         if (!conf.contains("reactiveinflux.url")) {
             conf.set("reactiveinflux.url", AppConf.getConfig("reactiveinflux.url"));
         }
+        if (sparkCassandraConnectionHost.nonEmpty) {
+            conf.set("spark.cassandra.connection.host", sparkCassandraConnectionHost.get.asInstanceOf[String])
+            println("spark.cassandra.connection.host", conf.get("spark.cassandra.connection.host"))
+        }
+
         // $COVERAGE-ON$
         val sc = new SparkContext(conf);
         setS3Conf(sc);

diff --git a/...tics-job-driver/src/main/scala/org/ekstep/analytics/framework/driver/BatchJobDriver.scala b/...tics-job-driver/src/main/scala/org/ekstep/analytics/framework/driver/BatchJobDriver.scala
@@ -19,7 +19,8 @@ object BatchJobDriver {
     def process[T, R](config: JobConfig, models: List[IBatchModel[T, R]])(implicit mf: Manifest[T], mfr: Manifest[R], sc: SparkContext) {
         JobContext.parallelization = CommonUtil.getParallelization(config);
         if (null == sc) {
-            implicit val sc = CommonUtil.getSparkContext(JobContext.parallelization, config.appName.getOrElse(config.model));
+            val sparkCassandraConnectionHost = config.modelParams.getOrElse(Map()).get("sparkCassandraConnectionHost")
+            implicit val sc = CommonUtil.getSparkContext(JobContext.parallelization, config.appName.getOrElse(config.model), sparkCassandraConnectionHost);
             try {
                 _process(config, models);
             } finally {