TG-990 feat: Upgrade Spark version to 3.0

batch-models/pom.xml

@@ -87,7 +87,7 @@
 		<dependency>
 			<groupId>org.apache.hadoop</groupId>
 			<artifactId>hadoop-client</artifactId>
-			<version>2.7.3</version>
+			<version>2.7.4</version>
 			<scope>provided</scope>
 		</dependency>
 		<dependency>
@@ -138,24 +138,24 @@
 			<version>4.1.0</version>
 			<scope>test</scope>
 		</dependency>
-		<!-- -->
+		<!--
 		<dependency>
 			<groupId>optional</groupId>
 			<artifactId>optional_${scala.version}</artifactId>
 			<version>1.0</version>
 		</dependency>
-
+		-->
 		<dependency>
 			<groupId>com.datastax.spark</groupId>
 			<artifactId>spark-cassandra-connector_${scala.maj.version}</artifactId>
-			<version>2.5.1</version>
-		</dependency>
-		<dependency>
-			<groupId>org.scalanlp</groupId>
-			<artifactId>breeze_${scala.maj.version}</artifactId>
-			<version>0.11.2</version>
-			<scope>provided</scope>
+			<version>3.0.1</version>
 		</dependency>
+<!--		<dependency>-->
+<!--			<groupId>org.scalanlp</groupId>-->
+<!--			<artifactId>breeze_${scala.maj.version}</artifactId>-->
+<!--			<version>0.11.2</version>-->
+<!--			<scope>provided</scope>-->
+<!--		</dependency>-->
 		<dependency>
 			<groupId>org.apache.spark</groupId>
 			<artifactId>spark-mllib_${scala.maj.version}</artifactId>
@@ -164,7 +164,7 @@
 			<exclusions>
 				<exclusion>
 					<groupId>org.apache.spark</groupId>
-					<artifactId>spark-core_2.11</artifactId>
+					<artifactId>spark-core_${scala.maj.version}</artifactId>
 				</exclusion>
 			</exclusions>
 		</dependency>
@@ -208,29 +208,31 @@
 		</dependency>
 		<dependency>
 			<groupId>net.liftweb</groupId>
-			<artifactId>lift-json_2.11</artifactId>
+			<artifactId>lift-json_${scala.maj.version}</artifactId>
 			<version>3.0.1</version>
 		</dependency>
 		<dependency>
 			<groupId>com.flyberrycapital</groupId>
-			<artifactId>scala-slack_2.11</artifactId>
-			<version>0.1.0</version>
+			<!--<artifactId>scala-slack_2.11</artifactId>
+			<version>0.1.0</version> -->
+			<artifactId>scala-slack_${scala.maj.version}</artifactId>
+			<version>0.3.1</version>
 		</dependency>
 		<dependency>
             <groupId>com.github.java-json-tools</groupId>
             <artifactId>json-schema-validator</artifactId>
             <version>2.2.8</version>
             <scope>test</scope>
         </dependency>
-		<dependency>
+		<!--<dependency>
 			<groupId>org.sunbird</groupId>
 			<artifactId>video-streaming</artifactId>
             <version>1.0</version>
-        </dependency>
+        </dependency>-->
         <dependency>
 			<groupId>org.sunbird</groupId>
-			<artifactId>cloud-store-sdk</artifactId>
-			<version>1.2.6</version>
+			<artifactId>cloud-store-sdk_${scala.maj.version}</artifactId>
+			<version>1.3.0</version>
 			<exclusions>
 				<exclusion>
 					<groupId>com.microsoft.azure</groupId>
@@ -240,24 +242,42 @@
                     <groupId>com.fasterxml.jackson.core</groupId>
                     <artifactId>jackson-core</artifactId>
                 </exclusion>
+				<exclusion>
+					<groupId>com.fasterxml.jackson.module</groupId>
+					<artifactId>jackson-module-scala_${scala.maj.version}.0-RC1</artifactId>
+				</exclusion>
 				<exclusion>
 					<groupId>org.apache.httpcomponents</groupId>
 					<artifactId>httpclient</artifactId>
 				</exclusion>
+				<exclusion>
+					<groupId>org.apache.hadoop</groupId>
+					<artifactId>hadoop-client</artifactId>
+				</exclusion>
 			</exclusions>
 		</dependency>
 		<dependency>
 			<groupId>com.microsoft.azure</groupId>
 			<artifactId>azure-storage</artifactId>
 			<version>3.0.0</version>
+			<exclusions>
+				<exclusion>
+					<groupId>org.apache.commons</groupId>
+					<artifactId>commons-lang3</artifactId>
+				</exclusion>
+				<exclusion>
+					<groupId>com.fasterxml.jackson.core</groupId>
+					<artifactId>jackson-core</artifactId>
+				</exclusion>
+			</exclusions>
 		</dependency>
 
 		<dependency>
 			<groupId>ing.wbaa.druid</groupId>
 			<artifactId>scruid_${scala.maj.version}</artifactId>
 			<version>2.4.0</version>
 			<scope>system</scope>
-			<systemPath>${project.basedir}/lib/scruid_2.11-2.4.0.jar</systemPath>
+			<systemPath>${project.basedir}/lib/scruid_2.12-2.4.0.jar</systemPath>
 		</dependency>
 		<dependency>
 		    <groupId>pl.allegro.tech</groupId>

batch-models/src/main/scala/org/ekstep/analytics/exhaust/BaseReportsJob.scala

@@ -0,0 +1,70 @@
+package org.ekstep.analytics.exhaust
+
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.types.StructType
+import org.ekstep.analytics.framework.{FrameworkContext, JobConfig, JobContext, StorageConfig}
+import org.ekstep.analytics.framework.util.CommonUtil
+import org.sunbird.cloud.storage.conf.AppConf
+
+trait BaseReportsJob {
+
+  def getReportingFrameworkContext()(implicit fc: Option[FrameworkContext]): FrameworkContext = {
+    fc match {
+      case Some(value) => {
+        value
+      }
+      case None => {
+        new FrameworkContext();
+      }
+    }
+  }
+
+  def openSparkSession(config: JobConfig): SparkSession = {
+
+    val modelParams = config.modelParams.getOrElse(Map[String, Option[AnyRef]]());
+    val sparkCassandraConnectionHost = modelParams.get("sparkCassandraConnectionHost")
+    val sparkElasticsearchConnectionHost = modelParams.get("sparkElasticsearchConnectionHost")
+    val sparkRedisConnectionHost = modelParams.get("sparkRedisConnectionHost")
+    val sparkUserDbRedisIndex = modelParams.get("sparkUserDbRedisIndex")
+    val sparkUserDbRedisPort = modelParams.get("sparkUserDbRedisPort")
+    JobContext.parallelization = CommonUtil.getParallelization(config)
+    val readConsistencyLevel = modelParams.getOrElse("cassandraReadConsistency", "LOCAL_QUORUM").asInstanceOf[String];
+    val writeConsistencyLevel = modelParams.getOrElse("cassandraWriteConsistency", "LOCAL_QUORUM").asInstanceOf[String]
+    val sparkSession = CommonUtil.getSparkSession(JobContext.parallelization, config.appName.getOrElse(config.model), sparkCassandraConnectionHost, sparkElasticsearchConnectionHost, Option(readConsistencyLevel),sparkRedisConnectionHost, sparkUserDbRedisIndex, sparkUserDbRedisPort)
+    setReportsStorageConfiguration(config)(sparkSession)
+    sparkSession;
+
+  }
+
+  def setReportsStorageConfiguration(config: JobConfig)(implicit spark: SparkSession) {
+
+    val modelParams = config.modelParams.getOrElse(Map[String, Option[AnyRef]]());
+    val store = modelParams.getOrElse("store", "local").asInstanceOf[String];
+    val storageKey = modelParams.getOrElse("storageKeyConfig", "reports_storage_key").asInstanceOf[String];
+    val storageSecret = modelParams.getOrElse("storageSecretConfig", "reports_storage_secret").asInstanceOf[String];
+    store.toLowerCase() match {
+      case "s3" =>
+        spark.sparkContext.hadoopConfiguration.set("fs.s3n.awsAccessKeyId", AppConf.getConfig(storageKey));
+        spark.sparkContext.hadoopConfiguration.set("fs.s3n.awsSecretAccessKey", AppConf.getConfig(storageSecret));
+      case "azure" =>
+        val storageKeyValue = AppConf.getConfig(storageKey);
+        spark.sparkContext.hadoopConfiguration.set("fs.azure", "org.apache.hadoop.fs.azure.NativeAzureFileSystem")
+        spark.sparkContext.hadoopConfiguration.set(s"fs.azure.account.key.$storageKeyValue.blob.core.windows.net", AppConf.getConfig(storageSecret))
+        spark.sparkContext.hadoopConfiguration.set(s"fs.azure.account.keyprovider.$storageKeyValue.blob.core.windows.net", "org.apache.hadoop.fs.azure.SimpleKeyProvider")
+      case _ =>
+
+    }
+
+  }
+
+  def getStorageConfig(config: JobConfig, key: String): StorageConfig = {
+
+    val modelParams = config.modelParams.getOrElse(Map[String, Option[AnyRef]]());
+    val container = modelParams.getOrElse("storageContainer", "reports").asInstanceOf[String]
+    val storageKey = modelParams.getOrElse("storageKeyConfig", "reports_storage_key").asInstanceOf[String];
+    val storageSecret = modelParams.getOrElse("storageSecretConfig", "reports_storage_secret").asInstanceOf[String];
+    val store = modelParams.getOrElse("store", "local").asInstanceOf[String]
+    StorageConfig(store, container, key, Option(storageKey), Option(storageSecret));
+  }
+
+}