[PIO-176] Hack in build.sbt for switching between Spark 1.x and 2.x should be cleaned up.
Closes #477
diff --git a/build.sbt b/build.sbt
index 9eea3e0..533fbef 100644
--- a/build.sbt
+++ b/build.sbt
@@ -129,8 +129,6 @@
settings(commonSettings: _*).
settings(commonTestSettings: _*).
enablePlugins(GenJavadocPlugin).
- settings(unmanagedSourceDirectories in Compile +=
- sourceDirectory.value / s"main/spark-${majorVersion(sparkVersion.value)}").
disablePlugins(sbtassembly.AssemblyPlugin)
val core = (project in file("core")).
diff --git a/data/src/main/spark-2/org/apache/predictionio/data/store/python/PPythonEventStore.scala b/data/src/main/scala/org/apache/predictionio/data/store/python/PPythonEventStore.scala
similarity index 100%
rename from data/src/main/spark-2/org/apache/predictionio/data/store/python/PPythonEventStore.scala
rename to data/src/main/scala/org/apache/predictionio/data/store/python/PPythonEventStore.scala
diff --git a/data/src/main/scala/org/apache/predictionio/data/view/DataView.scala b/data/src/main/scala/org/apache/predictionio/data/view/DataView.scala
index 1c47e10..ca92e8f 100644
--- a/data/src/main/scala/org/apache/predictionio/data/view/DataView.scala
+++ b/data/src/main/scala/org/apache/predictionio/data/view/DataView.scala
@@ -20,14 +20,10 @@
import org.apache.predictionio.annotation.Experimental
import org.apache.predictionio.data.storage.Event
-import org.apache.predictionio.data.SparkVersionDependent
-
import grizzled.slf4j.Logger
import org.apache.predictionio.data.store.PEventStore
-
import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.DataFrame
-import org.apache.spark.sql.SaveMode
+import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import org.apache.spark.SparkContext
import org.joda.time.DateTime
@@ -52,7 +48,6 @@
* @param name identify the DataFrame created
* @param version used to track changes to the conversionFunction, e.g. version = "20150413"
* and update whenever the function is changed.
- * @param sqlContext SQL context
* @tparam E the output type of the conversion function. The type needs to extend Product
* (e.g. case class)
* @return a DataFrame of events
@@ -69,7 +64,7 @@
@transient lazy val logger = Logger[this.type]
- val sqlSession = SparkVersionDependent.sqlSession(sc)
+ val sqlSession = SparkSession.builder().getOrCreate()
val beginTime = startTime match {
case Some(t) => t
diff --git a/data/src/main/spark-2/org/apache/predictionio/data/SparkVersionDependent.scala b/data/src/main/spark-2/org/apache/predictionio/data/SparkVersionDependent.scala
deleted file mode 100644
index 3d07bdf..0000000
--- a/data/src/main/spark-2/org/apache/predictionio/data/SparkVersionDependent.scala
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.
- */
-
-
-package org.apache.predictionio.data
-
-import org.apache.spark.SparkContext
-import org.apache.spark.sql.SparkSession
-
-object SparkVersionDependent {
-
- def sqlSession(sc: SparkContext): SparkSession = {
- SparkSession.builder().getOrCreate()
- }
-
-}
diff --git a/storage/jdbc/src/main/scala/org/apache/predictionio/data/storage/jdbc/JDBCPEvents.scala b/storage/jdbc/src/main/scala/org/apache/predictionio/data/storage/jdbc/JDBCPEvents.scala
index d31e592..4fa8b9f 100644
--- a/storage/jdbc/src/main/scala/org/apache/predictionio/data/storage/jdbc/JDBCPEvents.scala
+++ b/storage/jdbc/src/main/scala/org/apache/predictionio/data/storage/jdbc/JDBCPEvents.scala
@@ -20,12 +20,10 @@
import java.sql.{DriverManager, ResultSet}
import com.github.nscala_time.time.Imports._
-import org.apache.predictionio.data.storage.{
- DataMap, Event, PEvents, StorageClientConfig}
-import org.apache.predictionio.data.SparkVersionDependent
+import org.apache.predictionio.data.storage.{DataMap, Event, PEvents, StorageClientConfig}
import org.apache.spark.SparkContext
import org.apache.spark.rdd.{JdbcRDD, RDD}
-import org.apache.spark.sql.SaveMode
+import org.apache.spark.sql.{SaveMode, SparkSession}
import org.json4s.JObject
import org.json4s.native.Serialization
import scalikejdbc._
@@ -121,7 +119,7 @@
}
def write(events: RDD[Event], appId: Int, channelId: Option[Int])(sc: SparkContext): Unit = {
- val sqlSession = SparkVersionDependent.sqlSession(sc)
+ val sqlSession = SparkSession.builder().getOrCreate()
import sqlSession.implicits._
val tableName = JDBCUtils.eventTableName(namespace, appId, channelId)
diff --git a/tools/src/main/scala/org/apache/predictionio/tools/export/EventsToFile.scala b/tools/src/main/scala/org/apache/predictionio/tools/export/EventsToFile.scala
index 0372a44..9b6dbb5 100644
--- a/tools/src/main/scala/org/apache/predictionio/tools/export/EventsToFile.scala
+++ b/tools/src/main/scala/org/apache/predictionio/tools/export/EventsToFile.scala
@@ -21,14 +21,12 @@
import org.apache.predictionio.controller.Utils
import org.apache.predictionio.data.storage.EventJson4sSupport
import org.apache.predictionio.data.storage.Storage
-import org.apache.predictionio.data.SparkVersionDependent
import org.apache.predictionio.tools.Runner
import org.apache.predictionio.workflow.WorkflowContext
import org.apache.predictionio.workflow.WorkflowUtils
import org.apache.predictionio.workflow.CleanupFunctions
-
import grizzled.slf4j.Logging
-import org.apache.spark.sql.SaveMode
+import org.apache.spark.sql.{SaveMode, SparkSession}
import org.json4s.native.Serialization._
case class EventsToFileArgs(
@@ -93,7 +91,7 @@
mode = "Export",
batch = "App ID " + args.appId + channelStr,
executorEnv = Runner.envStringToMap(args.env))
- val sqlSession = SparkVersionDependent.sqlSession(sc)
+ val sqlSession = SparkSession.builder().getOrCreate()
val events = Storage.getPEvents()
val eventsRdd = events.find(appId = args.appId, channelId = channelId)(sc)
val jsonStringRdd = eventsRdd.map(write(_))
