LIVY-230. SparkSession is not avalible for PySparkInterperter and SparkRInterperter

Closes  #210

Author: zjffdu

repl/src/main/resources/fake_shell.py

@@ -514,6 +514,7 @@ def main():
     sys.stdout = UnicodeDecodingStringIO()
     sys.stderr = UnicodeDecodingStringIO()
 
+    spark_major_version = os.getenv("LIVY_SPARK_MAJOR_VERSION")
     try:
         listening_port = 0
         if os.environ.get("LIVY_TEST") != "true":
@@ -523,6 +524,8 @@ def main():
             exec('from pyspark.sql import HiveContext', global_dict)
             exec('from pyspark.streaming import StreamingContext', global_dict)
             exec('import pyspark.cloudpickle as cloudpickle', global_dict)
+            if spark_major_version >= "2":
+                exec('from pyspark.shell import spark', global_dict)
 
             #Start py4j callback server
             from py4j.protocol import ENTRY_POINT_OBJECT_ID

repl/src/main/scala/com/cloudera/livy/repl/PythonInterpreter.scala

@@ -68,7 +68,7 @@ object PythonInterpreter extends Logging {
     env.put("PYTHONUNBUFFERED", "YES")
     env.put("PYSPARK_GATEWAY_PORT", "" + gatewayServer.getListeningPort)
     env.put("SPARK_HOME", sys.env.getOrElse("SPARK_HOME", "."))
-
+    env.put("LIVY_SPARK_MAJOR_VERSION", conf.get("spark.livy.spark_major_version", "1"))
     builder.redirectError(Redirect.PIPE)
     val process = builder.start()
     new PythonInterpreter(process, gatewayServer, kind.toString)

repl/src/main/scala/com/cloudera/livy/repl/SparkRInterpreter.scala

@@ -114,7 +114,8 @@ object SparkRInterpreter {
 
       builder.redirectError(Redirect.PIPE)
       val process = builder.start()
-      new SparkRInterpreter(process, backendInstance, backendThread)
+      new SparkRInterpreter(process, backendInstance, backendThread,
+        conf.get("spark.livy.spark_major_version", "1"))
     } catch {
       case e: Exception =>
         if (backendThread != null) {
@@ -125,9 +126,8 @@ object SparkRInterpreter {
   }
 }
 
-class SparkRInterpreter(process: Process, backendInstance: Any, backendThread: Thread)
-  extends ProcessInterpreter(process)
-{
+class SparkRInterpreter(process: Process, backendInstance: Any, backendThread: Thread,
+  sparkMajorVersion: String) extends ProcessInterpreter(process) {
   import SparkRInterpreter._
 
   implicit val formats = DefaultFormats
@@ -141,8 +141,16 @@ class SparkRInterpreter(process: Process, backendInstance: Any, backendThread: T
     sendRequest("options(error = dump.frames)")
     if (!ClientConf.TEST_MODE) {
       sendRequest("library(SparkR)")
-      sendRequest("sc <- sparkR.init()")
-      sendRequest("sqlContext <- sparkRSQL.init(sc)")
+
+      if (sparkMajorVersion >= "2") {
+        sendRequest("spark <- SparkR::sparkR.session()")
+        sendRequest(
+          """sc <- SparkR:::callJStatic("org.apache.spark.sql.api.r.SQLUtils",
+            "getJavaSparkContext", spark)""")
+      } else {
+        sendRequest("sc <- sparkR.init()")
+        sendRequest("sqlContext <- sparkRSQL.init(sc)")
+      }
     }
 
     isStarted.countDown()

server/src/main/scala/com/cloudera/livy/server/interactive/InteractiveSession.scala

@@ -125,7 +125,8 @@ class InteractiveSession(
     val enableHiveContext = livyConf.getBoolean(LivyConf.ENABLE_HIVE_CONTEXT)
     val sparkMajorVersion =
       LivySparkUtils.formatSparkVersion(LivySparkUtils.sparkSubmitVersion(livyConf))._1
-
+    // pass spark.livy.spark_major_version to driver
+    builderProperties.put("spark.livy.spark_major_version", sparkMajorVersion.toString)
     if (sparkMajorVersion <= 1) {
       builderProperties.put("spark.repl.enableHiveContext",
         livyConf.getBoolean(LivyConf.ENABLE_HIVE_CONTEXT).toString)