Add support for Spark 2.2

Author: Jaroslaw Osmanski

build/build_and_run_tests.sh

@@ -32,6 +32,7 @@ run_tests() {
   sbt clean
   sbt -DSPARK_VERSION=$1 test ds-it
 }
+run_tests 2.2.0
 run_tests 2.1.1
 run_tests 2.0.2
 

build/workflow-executor_build.sh

@@ -29,3 +29,4 @@ build_workflowexecutor "2.0.1"
 build_workflowexecutor "2.0.2"
 build_workflowexecutor "2.1.0"
 build_workflowexecutor "2.1.1"
+build_workflowexecutor "2.2.0"

deployment/spark-docker/Dockerfile

@@ -25,7 +25,7 @@ ENV SPARK_PACKAGE spark-$SPARK_VERSION-bin-hadoop$HADOOP_VERSION
 ENV SPARK_HOME /opt/spark-$SPARK_VERSION
 ENV PATH $PATH:$SPARK_HOME/bin
 RUN wget -q -O - \
-  "http://d3kbcqa49mib13.cloudfront.net/$SPARK_PACKAGE.tgz" \
+  "https://archive.apache.org/dist/spark/spark-$SPARK_VERSION/$SPARK_PACKAGE.tgz" \
   | gunzip \
   | tar x -C /tmp/ \
   && mv /tmp/$SPARK_PACKAGE $SPARK_HOME \

project/Dependencies.scala

@@ -33,6 +33,7 @@ object Version {
 
   val spark = sys.props.getOrElse("SPARK_VERSION", "2.1.1")
   val (scala, hadoop, akka, sprayRoutingLib) = spark match {
+    case "2.2.0"  => ("2.11.8", "2.7", "2.4.9", "routing-shapeless2")
     case "2.1.0" | "2.1.1" => ("2.11.8", "2.7", "2.4.9", "routing-shapeless2")
     case "2.0.0" | "2.0.1" | "2.0.2" => ("2.11.8", "2.7", "2.4.9", "routing")
   }

remote_notebook/code/sparkr_kernel/kernel_init.R

@@ -16,7 +16,7 @@ assign("sc", get(".sparkRjsc", envir = SparkR:::.sparkREnv), envir = .GlobalEnv)
 sparkSQLSession <- SparkR:::callJMethod(entryPoint, "getNewSparkSQLSession")
 
 sparkVersion <- SparkR:::callJMethod(sc, "version")
-if (sparkVersion %in% c("2.0.0", "2.0.1", "2.0.2", "2.1.0", "2.1.1")) {
+if (sparkVersion %in% c("2.0.0", "2.0.1", "2.0.2", "2.1.0", "2.1.1", "2.2.0", "2.2.1")) {
   assign(".sparkRsession", SparkR:::callJMethod(sparkSQLSession, "getSparkSession"), envir = SparkR:::.sparkREnv)
   assign("spark", get(".sparkRsession", envir = SparkR:::.sparkREnv), envir = .GlobalEnv)
 } else {

seahorse-workflow-executor/build.sbt

@@ -22,23 +22,51 @@ lazy val settingsForNotPublished = CommonSettingsPlugin.assemblySettings ++
 
 lazy val sparkVersion = Version.spark
 
+
+
 lazy val sparkUtils = sparkVersion match {
   case "2.0.0" | "2.0.1" | "2.0.2" =>
     val sparkUtils2_0_x = project in file("sparkutils2.0.x") settings settingsForPublished
     sparkUtils2_0_x
   case "2.1.0" | "2.1.1" =>
     val sparkUtils2_1_0 = project in file("sparkutils2.1.x") settings settingsForPublished
     sparkUtils2_1_0
+  case "2.2.0" =>
+    val sparkUtils2_1_0 = project in file("sparkutils2.2.x") settings settingsForPublished
+    sparkUtils2_1_0
 }
 
-lazy val sparkUtils2_x = project in file(s"sparkutils2.x") dependsOn sparkUtils settings settingsForPublished
+lazy val sparkUtils2_x = project in file(s"sparkutils2.x") dependsOn (csvlib, sparkUtils) settings settingsForPublished
+
+lazy val csv2_2 = project in file(s"sparkutilsfeatures/csv2_2") settings settingsForPublished
+lazy val csv2_0 = project in file(s"sparkutilsfeatures/csv2_0") dependsOn sparkUtils settings settingsForPublished
+
+lazy val csvlib = sparkVersion match {
+  case "2.0.0" | "2.0.1" | "2.0.2" =>
+    csv2_0
+  case "2.1.0" | "2.1.1" =>
+    csv2_0
+  case "2.2.0" =>
+    csv2_2
+}
+
+lazy val readjsondataset = project in file(s"sparkutilsfeatures/readjsondataset") dependsOn sparkUtils2_x settings settingsForPublished
+lazy val readjsondataframe = project in file(s"sparkutilsfeatures/readjsondataframe") dependsOn sparkUtils2_x settings settingsForPublished
+
+lazy val readjson = sparkVersion match {
+  case "2.0.0" | "2.0.1" | "2.0.2" => readjsondataframe
+  case "2.1.0" | "2.1.1" => readjsondataframe
+  case "2.2.0" => readjsondataset
+}
 
 lazy val rootProject = project
   .in(file("."))
   .settings(name := "seahorse")
   .settings(PublishSettings.disablePublishing)
   .aggregate(
     api,
+    csvlib,
+    readjson,
     sparkUtils2_x,
     sparkUtils,
     commons,
@@ -54,7 +82,7 @@ lazy val api = project settings settingsForPublished
 
 lazy val commons = project dependsOn (api, sparkUtils2_x) settings settingsForPublished
 
-lazy val deeplang = project dependsOn (commons,
+lazy val deeplang = project dependsOn (commons, readjson, csvlib,
 commons % "test->test",
 graph,
 graph % "test->test",

seahorse-workflow-executor/build/build_and_run_tests.sh

@@ -15,8 +15,8 @@
 
 
 TASKS="clean scalastyle test:scalastyle test it:compile ds-it"
-#sbt -DSPARK_VERSION=2.0.0 $TASKS
-#sbt -DSPARK_VERSION=2.0.1 $TASKS
-#sbt -DSPARK_VERSION=2.1.0 $TASKS
+sbt -DSPARK_VERSION=2.0.0 $TASKS
+sbt -DSPARK_VERSION=2.0.1 $TASKS
+sbt -DSPARK_VERSION=2.1.0 $TASKS
 sbt -DSPARK_VERSION=2.1.1 $TASKS
-sbt -DSPARK_VERSION=2.0.2 $TASKS
+sbt -DSPARK_VERSION=2.2.0 $TASKS

seahorse-workflow-executor/deeplang/src/main/scala/ai/deepsense/deeplang/doperations/readwritedataframe/filestorage/DriverFiles.scala

@@ -19,18 +19,20 @@ package ai.deepsense.deeplang.doperations.readwritedataframe.filestorage
 import java.io.{File, IOException, PrintWriter}
 
 import scala.io.Source
+
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.execution.datasources.csv.{DataframeToDriverCsvFileWriter, RawCsvRDDToDataframe}
-import org.apache.spark.sql.{SaveMode, DataFrame => SparkDataFrame}
+import org.apache.spark.sql.{Dataset, Encoders, Row, SaveMode, DataFrame => SparkDataFrame}
 import ai.deepsense.commons.resources.ManagedResource
 import ai.deepsense.deeplang.ExecutionContext
 import ai.deepsense.deeplang.doperables.dataframe.DataFrame
 import ai.deepsense.deeplang.doperations.inout.{InputFileFormatChoice, OutputFileFormatChoice}
 import ai.deepsense.deeplang.doperations.readwritedataframe.filestorage.csv.CsvOptions
 import ai.deepsense.deeplang.doperations.readwritedataframe.{FilePath, FileScheme}
+import ai.deepsense.deeplang.readjsondataset.JsonReader
 import ai.deepsense.sparkutils.SQL
 
-object DriverFiles {
+object DriverFiles extends JsonReader {
 
   def read(driverPath: String, fileFormat: InputFileFormatChoice)
           (implicit context: ExecutionContext): SparkDataFrame = fileFormat match {
@@ -59,13 +61,14 @@ object DriverFiles {
     val lines = Source.fromFile(driverPath).getLines().toStream
     val fileLinesRdd = context.sparkContext.parallelize(lines)
 
-    RawCsvRDDToDataframe.parse(fileLinesRdd, context.sparkSQLSession, params)
+    RawCsvRDDToDataframe.parse(fileLinesRdd, context.sparkSQLSession.sparkSession, params)
   }
 
   private def readJson(driverPath: String)(implicit context: ExecutionContext) = {
     val lines = Source.fromFile(driverPath).getLines().toStream
     val fileLinesRdd = context.sparkContext.parallelize(lines)
-    context.sparkSQLSession.read.json(fileLinesRdd)
+    val sparkSession = context.sparkSQLSession.sparkSession
+    readJsonFromRdd(fileLinesRdd, sparkSession)
   }
 
   private def writeCsv
@@ -77,7 +80,8 @@ object DriverFiles {
       dataFrame.sparkDataFrame,
       params,
       dataFrame.schema.get,
-      path.pathWithoutScheme
+      path.pathWithoutScheme,
+      context.sparkSQLSession.sparkSession
     )
   }
 

seahorse-workflow-executor/docker/spark-standalone-cluster-manage.sh

@@ -55,7 +55,7 @@ function networkRm {
 ACTION=$1
 SPARK_VERSION=$2
 export SPARK_VERSION=${SPARK_VERSION}
-if  [ "$SPARK_VERSION" == "2.1.0" ] || [ "$SPARK_VERSION" == "2.1.1" ]; then
+if  [ "$SPARK_VERSION" == "2.1.0" ] || [ "$SPARK_VERSION" == "2.1.1" ] || [ "$SPARK_VERSION" == "2.2.0" ]; then
   export HADOOP_VERSION="2.7"
   # We use 2.7.1 for Spark 2.1.x despite the fact that the latter depends on 2.7.3, but 2.7.3
   # doesn't have docker image released yet.

seahorse-workflow-executor/project/Dependencies.scala

@@ -20,6 +20,7 @@ object Version {
 
   val spark = sys.props.getOrElse("SPARK_VERSION", "2.1.1")
   val (scala, java, hadoop, akka, apacheCommons) = spark match {
+    case "2.2.0" => ("2.11.8", "1.8", "2.7.3", "2.4.9", "3.5")
     case "2.1.0" | "2.1.1" => ("2.11.8", "1.8", "2.7.3", "2.4.9", "3.5")
     case "2.0.0" | "2.0.1" | "2.0.2" => ("2.11.8", "1.8", "2.7.1", "2.4.9", "3.3.+")
   }