[WIP] Add Inputer as DOperation

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/e2e_tests.sh

@@ -70,7 +70,7 @@ SPARK_STANDALONE_MANAGEMENT="./seahorse-workflow-executor/docker/spark-standalon
 MESOS_SPARK_DOCKER_COMPOSE="testing/mesos-spark-cluster/mesos-cluster.dc.yml"
 YARN_SPARK_DOCKER_COMPOSE="testing/yarn-spark-cluster/yarn-cluster.dc.yml"
 
-SPARK_VERSION="2.1.1"
+SPARK_VERSION="2.2.0"
 HADOOP_VERSION="2.7"
 
 ## Make sure that when job is aborted/killed all dockers will be turned off

build/manage-docker.py

@@ -31,7 +31,7 @@
 sbt_type = 'sbt'
 
 
-spark_version = "2.1.1"
+spark_version = "2.2.0"
 hadoop_version = "2.7"
 
 

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

@@ -31,8 +31,9 @@ object Version {
   val scoverage = "1.0.4"
   val slick = "3.1.1"
 
-  val spark = sys.props.getOrElse("SPARK_VERSION", "2.1.1")
+  val spark = sys.props.getOrElse("SPARK_VERSION", "2.2.0")
   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/CatalogRecorder.scala

@@ -188,6 +188,7 @@ class CatalogRecorder private (jars: Seq[URL]) {
     catalog.registerDOperation(DOperationCategories.Filtering, () => new FilterRows())
     catalog.registerDOperation(DOperationCategories.Filtering, () => new HandleMissingValues())
     catalog.registerDOperation(DOperationCategories.Filtering, () => new Projection())
+    catalog.registerDOperation(DOperationCategories.Filtering, () => new Imputer())
 
     catalog.registerDOperation(DOperationCategories.Transformation.Custom, () => new CreateCustomTransformer())
     catalog.registerDOperation(DOperationCategories.Transformation.Custom, () => new SqlColumnTransformation())

seahorse-workflow-executor/deeplang/src/main/scala/ai/deepsense/deeplang/doperations/Imputer.scala

@@ -0,0 +1,99 @@
+/**
+  * Copyright (c) 2016, CodiLime Inc.
+  */
+package ai.deepsense.deeplang.doperations
+
+import scala.language.reflectiveCalls
+import scala.reflect.runtime.universe.TypeTag
+
+import ai.deepsense.commons.models.Id
+import ai.deepsense.commons.utils.Version
+import ai.deepsense.deeplang._
+import ai.deepsense.deeplang.documentation.OperationDocumentation
+import ai.deepsense.deeplang.doperables.Transformer
+import ai.deepsense.deeplang.doperables.dataframe.DataFrame
+import ai.deepsense.deeplang.doperations.InputerTransformer.{Mean, Strategy}
+import ai.deepsense.deeplang.params.choice.Choice
+import ai.deepsense.deeplang.params.{ColumnSelectorParam, Param}
+import ai.deepsense.deeplang.params.selections.{MultipleColumnSelection, NameColumnSelection}
+import ai.deepsense.deeplang.params.wrappers.spark.{ChoiceParamWrapper, ParamsWithSparkWrappers}
+import org.apache.spark.ml
+import org.apache.spark.ml.feature.{Imputer => SparkImputer}
+import org.apache.spark.sql.types.StructType
+
+
+
+object InputerTransformer {
+  sealed abstract class Strategy(override val name: String) extends Choice {
+    override val params: Array[Param[_]] = Array()
+
+    override val choiceOrder: List[Class[_ <: Choice]] = List(
+      classOf[Mean],
+      classOf[Median]
+    )
+  }
+  case class Mean() extends Strategy("mean")
+  case class Median() extends Strategy("median")
+}
+
+
+class InputerTransformer extends Transformer with ParamsWithSparkWrappers {
+
+  val selectedColumns = ColumnSelectorParam(
+    name = "selected columns",
+    description = Some("Columns to complete missing values in."),
+    portIndex = 0)
+
+  def getSelectedColumns: MultipleColumnSelection = $(selectedColumns)
+
+  def setSelectedColumns(value: MultipleColumnSelection): this.type =
+    set(selectedColumns, value)
+
+  def setSelectedColumns(retainedColumns: Seq[String]): this.type =
+    setSelectedColumns(
+      MultipleColumnSelection(
+        Vector(NameColumnSelection(retainedColumns.toSet)),
+        excluding = false))
+
+  val strategy = new ChoiceParamWrapper[
+    ml.param.Params { val strategy: ml.param.Param[String] }, Strategy](
+    name = "strategy",
+    description = Some("Algorithm used to compute missing value"),
+    sparkParamGetter = _.strategy)
+  setDefault(strategy, Mean())
+
+  override val params: Array[Param[_]] = Array(selectedColumns, strategy)
+
+
+
+  override def applyTransform(ctx: ExecutionContext, df: DataFrame): DataFrame = {
+    val columns = df.getColumnNames(getSelectedColumns)
+    val inputer = new SparkImputer()
+      .setInputCols(columns.toArray)
+      .setOutputCols(columns.toArray)
+    val sparkParams =
+      sparkParamMap(inputer, df.sparkDataFrame.schema)
+
+    val imputerModel = inputer.fit(df.sparkDataFrame, sparkParams)
+    val res = imputerModel.transform(df.sparkDataFrame).toDF()
+    DataFrame.fromSparkDataFrame(res)
+  }
+
+  override def applyTransformSchema(schema: StructType): Option[StructType] = {
+    Some(schema)
+  }
+}
+
+
+class Imputer extends TransformerAsOperation[InputerTransformer] with OperationDocumentation {
+  override val id: Id = "37d8ce21-0aa9-4448-8fb0-defb58c5e53f"
+  override val name: String = "Imputer"
+  override val description: String =
+    "Completes missing values using Estimator on a DataFrame"
+
+  override lazy val tTagTI_0: TypeTag[DataFrame] = typeTag
+  override lazy val tTagTO_0: TypeTag[DataFrame] = typeTag
+  override lazy val tTagTO_1: TypeTag[InputerTransformer] = typeTag
+
+  override val since: Version = Version(1, 5, 0)
+}

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

@@ -18,8 +18,9 @@ import sbt._
 
 object Version {
 
-  val spark = sys.props.getOrElse("SPARK_VERSION", "2.1.1")
+  val spark = sys.props.getOrElse("SPARK_VERSION", "2.2.0")
   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.+")
   }

seahorse-workflow-executor/python/pyexecutor/code_executor.py

@@ -91,7 +91,7 @@ def _run_custom_code(self, workflow_id, node_id, custom_operation_code):
 
         new_sql_context = None
         spark_version = self.spark_context.version
-        if spark_version in ["2.0.0", "2.0.1", "2.0.2", "2.1.0", "2.1.1"]:
+        if spark_version in ["2.0.0", "2.0.1", "2.0.2", "2.1.0", "2.1.1", "2.2.0"]:
             new_sql_context = SQLContext(self.spark_context, new_spark_session)
         else:
             raise ValueError("Spark version {} is not supported".format(spark_version))

seahorse-workflow-executor/python/pyexecutor/pyexecutor.py

@@ -71,7 +71,7 @@ def _initialize_spark_contexts(gateway):
         java_spark_sql_session = gateway.entry_point.getSparkSQLSession()
         spark_version = spark_context.version
         spark_sql_session = None
-        if spark_version in ["2.0.0", "2.0.1", "2.0.2", "2.1.0", "2.1.1"]:
+        if spark_version in ["2.0.0", "2.0.1", "2.0.2", "2.1.0", "2.1.1", "2.2.0"]:
             from pyspark.sql import SparkSession
             java_spark_session = java_spark_sql_session.getSparkSession()
             spark_sql_session = SparkSession(spark_context, java_spark_session)

seahorse-workflow-executor/scalastyle-config.xml

@@ -22,7 +22,7 @@
     <parameters>
       <parameter name="regex">true</parameter>
       <parameter name="header"><![CDATA[/\*\*
- \* Copyright 201[567] deepsense.ai \(CodiLime, Inc\)
+ \* Copyright 201[56789] deepsense.ai \(CodiLime, Inc\)
  \*
  \* Licensed under the Apache License, Version 2.0 \(the \"License\"\);
  \* you may not use this file except in compliance with the License\.

seahorse-workflow-executor/sparkutils2.1.x/src/main/scala/ai/deepsense/sparkutils/readwritedataframe/SparkCsvReader.scala

@@ -14,19 +14,18 @@
  * limitations under the License.
  */
 
+
 package org.apache.spark.sql.execution.datasources.csv
 
 import org.apache.spark.rdd.RDD
-import org.apache.spark.sql._
-import org.apache.spark.sql.execution.LogicalRDD
-import org.apache.spark.sql.types.{StructType, _}
+
 
 object SparkCsvReader {
   def create(options: CSVOptions): CsvReader = new CsvReader(options)
   def univocityTokenizer(
-      rdd: RDD[String],
-      header: Array[String],
-      firstLine: String,
-      options: CSVOptions): RDD[Array[String]] =
+    rdd: RDD[String],
+    header: Array[String],
+    firstLine: String,
+    options: CSVOptions): RDD[Array[String]] =
     CSVRelation.univocityTokenizer(rdd, firstLine, options)
 }

seahorse-workflow-executor/sparkutils2.2.x/build.sbt

@@ -0,0 +1,19 @@
+/**
+ * Copyright 2017 deepsense.ai (CodiLime, Inc)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+name := "seahorse-executor-sparkutils2.2.x"
+
+libraryDependencies ++= Dependencies.sparkutils(Version.spark)