Repository: zeppelin
Updated Branches:
  refs/heads/branch-0.8 7ba5e6ffc -> 3745e65b4


[ZEPPELIN-2909]. Support shared SparkContext across language in livy interpreter

### What is this PR for?
LIVY-194 implement the shared SparkContext across languages, this ticket is 
trying to integrate this feature.

### What type of PR is it?
[ Feature ]

### Todos
* [ ] - Task

### What is the Jira issue?
* https://issues.apache.org/jira/browse/ZEPPELIN-2909

### How should this be tested?
Tested is added

### Screenshots (if appropriate)

### Questions:
* Does the licenses files need update? No
* Is there breaking changes for older versions? No
* Does this needs documentation? No

Author: Jeff Zhang <zjf...@apache.org>

Closes #2587 from zjffdu/ZEPPELIN-2909 and squashes the following commits:

d6e38e6 [Jeff Zhang] [ZEPPELIN-2909]. Support shared SparkContext across 
language in livy interpreter

(cherry picked from commit ad6e691812957e240d37918bbeae4a3c537fcccf)
Signed-off-by: Jeff Zhang <zjf...@apache.org>


Project: http://git-wip-us.apache.org/repos/asf/zeppelin/repo
Commit: http://git-wip-us.apache.org/repos/asf/zeppelin/commit/3745e65b
Tree: http://git-wip-us.apache.org/repos/asf/zeppelin/tree/3745e65b
Diff: http://git-wip-us.apache.org/repos/asf/zeppelin/diff/3745e65b

Branch: refs/heads/branch-0.8
Commit: 3745e65b4ed8a678324dd34953ee7ca276ab8e34
Parents: 7ba5e6f
Author: Jeff Zhang <zjf...@apache.org>
Authored: Wed Aug 23 20:34:15 2017 +0800
Committer: Jeff Zhang <zjf...@apache.org>
Committed: Thu Feb 8 21:38:25 2018 +0800

----------------------------------------------------------------------
 .travis.yml                                     |   8 +-
 docs/interpreter/livy.md                        |   4 +
 livy/pom.xml                                    | 148 ++++++++++++++++++-
 .../zeppelin/livy/BaseLivyInterpreter.java      |  88 +++++++++--
 .../zeppelin/livy/LivySharedInterpreter.java    | 108 ++++++++++++++
 .../zeppelin/livy/LivySparkInterpreter.java     |   1 +
 .../zeppelin/livy/LivySparkSQLInterpreter.java  |  36 ++++-
 .../org/apache/zeppelin/livy/LivyVersion.java   |   5 +
 .../src/main/resources/interpreter-setting.json |  16 ++
 .../apache/zeppelin/livy/LivyInterpreterIT.java | 133 +++++++++++++++--
 10 files changed, 508 insertions(+), 39 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/zeppelin/blob/3745e65b/.travis.yml
----------------------------------------------------------------------
diff --git a/.travis.yml b/.travis.yml
index fd8c868..61aab1f 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -107,17 +107,17 @@ matrix:
       dist: trusty
       env: PYTHON="2" SCALA_VER="2.11" SPARK_VER="1.6.3" HADOOP_VER="2.6" 
PROFILE="-Pweb-ci -Pspark-1.6 -Phadoop3 -Phadoop-2.6 -Pscala-2.11" 
SPARKR="true" BUILD_FLAG="install -DskipTests -DskipRat" TEST_FLAG="test 
-DskipRat" MODULES="-pl 
.,zeppelin-interpreter,zeppelin-zengine,zeppelin-server,zeppelin-display,spark/interpreter,spark/scala-2.10,spark/scala-2.11,spark/spark-dependencies,python"
 TEST_PROJECTS="-Dtest=ZeppelinSparkClusterTest,org.apache.zeppelin.spark.* 
-DfailIfNoTests=false"
 
-    # Test python/pyspark with python 2, livy 0.2
+    # Test python/pyspark with python 2, livy 0.5
     - sudo: required
       dist: trusty
       jdk: "openjdk7"
-      env: PYTHON="2" SCALA_VER="2.10" SPARK_VER="1.6.1" HADOOP_VER="2.6" 
LIVY_VER="0.4.0-incubating" PROFILE="-Pspark-1.6 -Phadoop2 -Phadoop-2.6 
-Plivy-0.2 -Pscala-2.10" BUILD_FLAG="install -am -DskipTests -DskipRat" 
TEST_FLAG="verify -DskipRat" MODULES="-pl 
.,zeppelin-interpreter,zeppelin-display,spark/interpreter,spark/scala-2.10,spark/scala-2.11,spark/spark-dependencies,python,livy"
 
TEST_PROJECTS="-Dtest=LivySQLInterpreterTest,org.apache.zeppelin.spark.PySpark*Test,org.apache.zeppelin.python.*
 -Dpyspark.test.exclude='' -DfailIfNoTests=false"
+      env: PYTHON="2" SCALA_VER="2.10" SPARK_VER="1.6.3" HADOOP_VER="2.6" 
LIVY_VER="0.5.0-incubating" PROFILE="-Pspark-1.6 -Phadoop2 -Phadoop-2.6 
-Pscala-2.10" BUILD_FLAG="install -am -DskipTests -DskipRat" TEST_FLAG="verify 
-DskipRat" MODULES="-pl 
.,zeppelin-interpreter,zeppelin-display,spark/interpreter,spark/scala-2.10,spark/scala-2.11,spark/spark-dependencies,python,livy"
 
TEST_PROJECTS="-Dtest=LivySQLInterpreterTest,org.apache.zeppelin.spark.PySpark*Test,org.apache.zeppelin.python.*
 -Dpyspark.test.exclude='' -DfailIfNoTests=false"
 
-    # Test python/pyspark with python 3, livy 0.3
+    # Test python/pyspark with python 3, livy 0.5
     - sudo: required
       dist: trusty
       jdk: "openjdk7"
-      env: PYTHON="3" SCALA_VER="2.11" SPARK_VER="2.0.0" HADOOP_VER="2.6" 
LIVY_VER="0.4.0-incubating" PROFILE="-Pspark-2.0 -Phadoop3 -Phadoop-2.6 
-Pscala-2.11 -Plivy-0.3" BUILD_FLAG="install -am -DskipTests -DskipRat" 
TEST_FLAG="verify -DskipRat" MODULES="-pl 
.,zeppelin-interpreter,zeppelin-display,spark/interpreter,spark/scala-2.10,spark/scala-2.11,spark/spark-dependencies,python,livy"
 
TEST_PROJECTS="-Dtest=LivySQLInterpreterTest,org.apache.zeppelin.spark.PySpark*Test,org.apache.zeppelin.python.*
 -Dpyspark.test.exclude='' -DfailIfNoTests=false"
+      env: PYTHON="3" SCALA_VER="2.11" SPARK_VER="2.0.0" HADOOP_VER="2.6" 
LIVY_VER="0.5.0-incubating" PROFILE="-Pspark-2.0 -Phadoop3 -Phadoop-2.6 
-Pscala-2.11" BUILD_FLAG="install -am -DskipTests -DskipRat" TEST_FLAG="verify 
-DskipRat" MODULES="-pl 
.,zeppelin-interpreter,zeppelin-display,spark/interpreter,spark/scala-2.10,spark/scala-2.11,spark/spark-dependencies,python,livy"
 
TEST_PROJECTS="-Dtest=LivySQLInterpreterTest,org.apache.zeppelin.spark.PySpark*Test,org.apache.zeppelin.python.*
 -Dpyspark.test.exclude='' -DfailIfNoTests=false"
       
 before_install:
   # check files included in commit range, clear bower_components if a 
bower.json file has changed.

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/3745e65b/docs/interpreter/livy.md
----------------------------------------------------------------------
diff --git a/docs/interpreter/livy.md b/docs/interpreter/livy.md
index d53672a..e4784d4 100644
--- a/docs/interpreter/livy.md
+++ b/docs/interpreter/livy.md
@@ -216,6 +216,10 @@ select * from products where ${product_id=1}
 
 And creating dynamic formst programmatically is not feasible in livy 
interpreter, because ZeppelinContext is not available in livy interpreter.
 
+## Shared SparkContext
+Starting from livy 0.5 which is supported by Zeppelin 0.8.0, SparkContext is 
shared between scala, python, r and sql.
+That means you can query the table via `%livy.sql` when this table is 
registered in `%livy.spark`, `%livy.pyspark`, `$livy.sparkr`.
+
 ## FAQ
 
 Livy debugging: If you see any of these in error console

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/3745e65b/livy/pom.xml
----------------------------------------------------------------------
diff --git a/livy/pom.xml b/livy/pom.xml
index 0ec174a..d33adf6 100644
--- a/livy/pom.xml
+++ b/livy/pom.xml
@@ -42,8 +42,9 @@
         
<spring.security.kerberosclient>1.0.1.RELEASE</spring.security.kerberosclient>
 
         <!--test library versions-->
-        <livy.version>0.4.0-incubating</livy.version>
+        <livy.version>0.5.0-incubating</livy.version>
         <spark.version>2.1.0</spark.version>
+        <hadoop.version>2.6.0</hadoop.version>
         <!--plugin versions-->
         <plugin.failsafe.version>2.16</plugin.failsafe.version>
         <plugin.antrun.version>1.8</plugin.antrun.version>
@@ -105,6 +106,30 @@
                     <groupId>org.apache.spark</groupId>
                     <artifactId>spark-yarn_${scala.binary.version}</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-auth</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-common</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-hdfs</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-yarn-client</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-client</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-yarn-server-tests</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
         <dependency>
@@ -188,6 +213,127 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-auth</artifactId>
+            <version>${hadoop.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>${hadoop.version}</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.guava</groupId>
+                    <artifactId>guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <classifier>tests</classifier>
+            <version>${hadoop.version}</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.guava</groupId>
+                    <artifactId>guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-hdfs</artifactId>
+            <version>${hadoop.version}</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.netty</groupId>
+                    <artifactId>netty</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.google.guava</groupId>
+                    <artifactId>guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-hdfs</artifactId>
+            <classifier>tests</classifier>
+            <version>${hadoop.version}</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.netty</groupId>
+                    <artifactId>netty</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.google.guava</groupId>
+                    <artifactId>guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-client</artifactId>
+            <version>${hadoop.version}</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.guava</groupId>
+                    <artifactId>guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-yarn-client</artifactId>
+            <version>${hadoop.version}</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.guava</groupId>
+                    <artifactId>guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-yarn-api</artifactId>
+            <version>${hadoop.version}</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.guava</groupId>
+                    <artifactId>guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-yarn-server-tests</artifactId>
+            <classifier>tests</classifier>
+            <version>${hadoop.version}</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.guava</groupId>
+                    <artifactId>guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
     </dependencies>
 
     <build>

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/3745e65b/livy/src/main/java/org/apache/zeppelin/livy/BaseLivyInterpreter.java
----------------------------------------------------------------------
diff --git 
a/livy/src/main/java/org/apache/zeppelin/livy/BaseLivyInterpreter.java 
b/livy/src/main/java/org/apache/zeppelin/livy/BaseLivyInterpreter.java
index f3b7579..724a4b3 100644
--- a/livy/src/main/java/org/apache/zeppelin/livy/BaseLivyInterpreter.java
+++ b/livy/src/main/java/org/apache/zeppelin/livy/BaseLivyInterpreter.java
@@ -57,6 +57,8 @@ import org.apache.zeppelin.interpreter.InterpreterException;
 import org.apache.zeppelin.interpreter.InterpreterResult;
 import org.apache.zeppelin.interpreter.InterpreterResultMessage;
 import org.apache.zeppelin.interpreter.InterpreterUtils;
+import org.apache.zeppelin.interpreter.LazyOpenInterpreter;
+import org.apache.zeppelin.interpreter.WrappedInterpreter;
 import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -95,6 +97,9 @@ public abstract class BaseLivyInterpreter extends Interpreter 
{
   private RestTemplate restTemplate;
   private Map<String, String> customHeaders = new HashMap<>();
 
+  // delegate to sharedInterpreter when it is available
+  protected LivySharedInterpreter sharedInterpreter;
+
   Set<Object> paragraphsToCancel = Collections.newSetFromMap(
       new ConcurrentHashMap<Object, Boolean>());
   private ConcurrentHashMap<String, Integer> paragraphId2StmtProgressMap =
@@ -144,7 +149,13 @@ public abstract class BaseLivyInterpreter extends 
Interpreter {
   @Override
   public void open() throws InterpreterException {
     try {
-      initLivySession();
+      this.livyVersion = getLivyVersion();
+      if (this.livyVersion.isSharedSupported()) {
+        sharedInterpreter = getLivySharedInterpreter();
+      }
+      if (sharedInterpreter == null || !sharedInterpreter.isSupported()) {
+        initLivySession();
+      }
     } catch (LivyException e) {
       String msg = "Fail to create session, please check livy interpreter log 
and " +
           "livy server log";
@@ -152,8 +163,32 @@ public abstract class BaseLivyInterpreter extends 
Interpreter {
     }
   }
 
+  protected LivySharedInterpreter getLivySharedInterpreter() throws 
InterpreterException {
+    LazyOpenInterpreter lazy = null;
+    LivySharedInterpreter sharedInterpreter = null;
+    Interpreter p = getInterpreterInTheSameSessionByClassName(
+        LivySharedInterpreter.class.getName());
+
+    while (p instanceof WrappedInterpreter) {
+      if (p instanceof LazyOpenInterpreter) {
+        lazy = (LazyOpenInterpreter) p;
+      }
+      p = ((WrappedInterpreter) p).getInnerInterpreter();
+    }
+    sharedInterpreter = (LivySharedInterpreter) p;
+
+    if (lazy != null) {
+      lazy.open();
+    }
+    return sharedInterpreter;
+  }
+
   @Override
   public void close() {
+    if (sharedInterpreter != null && sharedInterpreter.isSupported()) {
+      sharedInterpreter.close();
+      return;
+    }
     if (sessionInfo != null) {
       closeSession(sessionInfo.id);
       // reset sessionInfo to null so that we won't close it twice.
@@ -181,14 +216,6 @@ public abstract class BaseLivyInterpreter extends 
Interpreter {
     } else {
       LOGGER.info("Create livy session successfully with sessionId: {}", 
this.sessionInfo.id);
     }
-    // check livy version
-    try {
-      this.livyVersion = getLivyVersion();
-      LOGGER.info("Use livy " + livyVersion);
-    } catch (APINotFoundException e) {
-      this.livyVersion = new LivyVersion("0.2.0");
-      LOGGER.info("Use livy 0.2.0");
-    }
   }
 
   protected abstract String extractAppId() throws LivyException;
@@ -196,17 +223,30 @@ public abstract class BaseLivyInterpreter extends 
Interpreter {
   protected abstract String extractWebUIAddress() throws LivyException;
 
   public SessionInfo getSessionInfo() {
+    if (sharedInterpreter != null && sharedInterpreter.isSupported()) {
+      return sharedInterpreter.getSessionInfo();
+    }
     return sessionInfo;
   }
 
+  public String getCodeType() {
+    if (getSessionKind().equalsIgnoreCase("pyspark3")) {
+      return "pyspark";
+    }
+    return getSessionKind();
+  }
+
   @Override
   public InterpreterResult interpret(String st, InterpreterContext context) {
+    if (sharedInterpreter != null && sharedInterpreter.isSupported()) {
+      return sharedInterpreter.interpret(st, getCodeType(), context);
+    }
     if (StringUtils.isEmpty(st)) {
       return new InterpreterResult(InterpreterResult.Code.SUCCESS, "");
     }
 
     try {
-      return interpret(st, context.getParagraphId(), this.displayAppInfo, 
true);
+      return interpret(st, null, context.getParagraphId(), 
this.displayAppInfo, true);
     } catch (LivyException e) {
       LOGGER.error("Fail to interpret:" + st, e);
       return new InterpreterResult(InterpreterResult.Code.ERROR,
@@ -245,6 +285,10 @@ public abstract class BaseLivyInterpreter extends 
Interpreter {
 
   @Override
   public void cancel(InterpreterContext context) {
+    if (sharedInterpreter != null && sharedInterpreter.isSupported()) {
+      sharedInterpreter.cancel(context);
+      return;
+    }
     paragraphsToCancel.add(context.getParagraphId());
     LOGGER.info("Added paragraph " + context.getParagraphId() + " for 
cancellation.");
   }
@@ -256,6 +300,10 @@ public abstract class BaseLivyInterpreter extends 
Interpreter {
 
   @Override
   public int getProgress(InterpreterContext context) {
+    if (sharedInterpreter != null && sharedInterpreter.isSupported()) {
+      return sharedInterpreter.getProgress(context);
+    }
+
     if (livyVersion.isGetProgressSupported()) {
       String paraId = context.getParagraphId();
       Integer progress = paragraphId2StmtProgressMap.get(paraId);
@@ -312,11 +360,20 @@ public abstract class BaseLivyInterpreter extends 
Interpreter {
                                      String paragraphId,
                                      boolean displayAppInfo,
                                      boolean appendSessionExpired) throws 
LivyException {
+    return interpret(code, sharedInterpreter.isSupported() ? getSessionKind() 
: null,
+        paragraphId, displayAppInfo, appendSessionExpired);
+  }
+
+  public InterpreterResult interpret(String code,
+                                     String codeType,
+                                     String paragraphId,
+                                     boolean displayAppInfo,
+                                     boolean appendSessionExpired) throws 
LivyException {
     StatementInfo stmtInfo = null;
     boolean sessionExpired = false;
     try {
       try {
-        stmtInfo = executeStatement(new ExecuteRequest(code));
+        stmtInfo = executeStatement(new ExecuteRequest(code, codeType));
       } catch (SessionNotFoundException e) {
         LOGGER.warn("Livy session {} is expired, new session will be 
created.", sessionInfo.id);
         sessionExpired = true;
@@ -328,7 +385,7 @@ public abstract class BaseLivyInterpreter extends 
Interpreter {
             initLivySession();
           }
         }
-        stmtInfo = executeStatement(new ExecuteRequest(code));
+        stmtInfo = executeStatement(new ExecuteRequest(code, codeType));
       }
 
       // pull the statement status
@@ -731,11 +788,12 @@ public abstract class BaseLivyInterpreter extends 
Interpreter {
     }
   }
 
-  private static class ExecuteRequest {
+  static class ExecuteRequest {
     public final String code;
-
-    public ExecuteRequest(String code) {
+    public final String kind;
+    public ExecuteRequest(String code, String kind) {
       this.code = code;
+      this.kind = kind;
     }
 
     public String toJson() {

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/3745e65b/livy/src/main/java/org/apache/zeppelin/livy/LivySharedInterpreter.java
----------------------------------------------------------------------
diff --git 
a/livy/src/main/java/org/apache/zeppelin/livy/LivySharedInterpreter.java 
b/livy/src/main/java/org/apache/zeppelin/livy/LivySharedInterpreter.java
new file mode 100644
index 0000000..77e288b
--- /dev/null
+++ b/livy/src/main/java/org/apache/zeppelin/livy/LivySharedInterpreter.java
@@ -0,0 +1,108 @@
+/*
+ * 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.zeppelin.livy;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.zeppelin.interpreter.InterpreterContext;
+import org.apache.zeppelin.interpreter.InterpreterException;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.apache.zeppelin.interpreter.InterpreterUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Properties;
+
+/**
+ * Livy Interpreter for shared kind which share SparkContext across 
spark/pyspark/r
+ */
+public class LivySharedInterpreter extends BaseLivyInterpreter {
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(LivySharedInterpreter.class);
+
+  private boolean isSupported = false;
+
+  public LivySharedInterpreter(Properties property) {
+    super(property);
+  }
+
+  @Override
+  public void open() throws InterpreterException {
+    try {
+      // check livy version
+      try {
+        this.livyVersion = getLivyVersion();
+        LOGGER.info("Use livy " + livyVersion);
+      } catch (APINotFoundException e) {
+        // assume it is livy 0.2.0 when livy doesn't support rest api of 
fetching version.
+        this.livyVersion = new LivyVersion("0.2.0");
+        LOGGER.info("Use livy 0.2.0");
+      }
+
+      if (livyVersion.isSharedSupported()) {
+        LOGGER.info("LivySharedInterpreter is supported.");
+        isSupported = true;
+        initLivySession();
+      } else {
+        LOGGER.info("LivySharedInterpreter is not supported.");
+        isSupported = false;
+      }
+    } catch (LivyException e) {
+      String msg = "Fail to create session, please check livy interpreter log 
and " +
+          "livy server log";
+      throw new InterpreterException(msg, e);
+    }
+  }
+
+  public boolean isSupported() {
+    return isSupported;
+  }
+
+  public InterpreterResult interpret(String st, String codeType, 
InterpreterContext context) {
+    if (StringUtils.isEmpty(st)) {
+      return new InterpreterResult(InterpreterResult.Code.SUCCESS, "");
+    }
+
+    try {
+      return interpret(st, codeType, context.getParagraphId(), 
this.displayAppInfo, true);
+    } catch (LivyException e) {
+      LOGGER.error("Fail to interpret:" + st, e);
+      return new InterpreterResult(InterpreterResult.Code.ERROR,
+          InterpreterUtils.getMostRelevantMessage(e));
+    }
+  }
+
+  @Override
+  public String getSessionKind() {
+    return "shared";
+  }
+
+  @Override
+  protected String extractAppId() throws LivyException {
+    return null;
+  }
+
+  @Override
+  protected String extractWebUIAddress() throws LivyException {
+    return null;
+  }
+
+  public static void main(String[] args) {
+    ExecuteRequest request = new ExecuteRequest("1+1", null);
+    System.out.println(request.toJson());
+  }
+}

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/3745e65b/livy/src/main/java/org/apache/zeppelin/livy/LivySparkInterpreter.java
----------------------------------------------------------------------
diff --git 
a/livy/src/main/java/org/apache/zeppelin/livy/LivySparkInterpreter.java 
b/livy/src/main/java/org/apache/zeppelin/livy/LivySparkInterpreter.java
index 606ef64..066d0da 100644
--- a/livy/src/main/java/org/apache/zeppelin/livy/LivySparkInterpreter.java
+++ b/livy/src/main/java/org/apache/zeppelin/livy/LivySparkInterpreter.java
@@ -44,6 +44,7 @@ public class LivySparkInterpreter extends BaseLivyInterpreter 
{
   protected String extractWebUIAddress() throws LivyException {
     interpret(
         "val 
webui=sc.getClass.getMethod(\"ui\").invoke(sc).asInstanceOf[Some[_]].get",
+        null,
         null, false, false);
     return extractStatementResult(
         interpret(

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/3745e65b/livy/src/main/java/org/apache/zeppelin/livy/LivySparkSQLInterpreter.java
----------------------------------------------------------------------
diff --git 
a/livy/src/main/java/org/apache/zeppelin/livy/LivySparkSQLInterpreter.java 
b/livy/src/main/java/org/apache/zeppelin/livy/LivySparkSQLInterpreter.java
index 7b2d7d6..2faa350 100644
--- a/livy/src/main/java/org/apache/zeppelin/livy/LivySparkSQLInterpreter.java
+++ b/livy/src/main/java/org/apache/zeppelin/livy/LivySparkSQLInterpreter.java
@@ -19,11 +19,14 @@ package org.apache.zeppelin.livy;
 
 import org.apache.commons.lang.StringUtils;
 import static org.apache.commons.lang.StringEscapeUtils.escapeJavaScript;
+import org.apache.zeppelin.display.GUI;
 import org.apache.zeppelin.interpreter.*;
 import org.apache.zeppelin.scheduler.Scheduler;
 import org.apache.zeppelin.scheduler.SchedulerFactory;
+import org.apache.zeppelin.user.AuthenticationInfo;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Properties;
 
@@ -39,6 +42,7 @@ public class LivySparkSQLInterpreter extends 
BaseLivyInterpreter {
       "zeppelin.livy.spark.sql.maxResult";
 
   private LivySparkInterpreter sparkInterpreter;
+  private String codeType = null;
 
   private boolean isSpark2 = false;
   private int maxResult = 1000;
@@ -64,7 +68,21 @@ public class LivySparkSQLInterpreter extends 
BaseLivyInterpreter {
     // As we don't know whether livyserver use spark2 or spark1, so we will 
detect SparkSession
     // to judge whether it is using spark2.
     try {
-      InterpreterResult result = sparkInterpreter.interpret("spark", null, 
false, false);
+      InterpreterContext context = new InterpreterContext(
+          "noteId",
+          "paragraphId",
+          "replName",
+          "paragraphTitle",
+          "paragraphText",
+          new AuthenticationInfo(),
+          new HashMap<String, Object>(),
+          new GUI(),
+          new GUI(),
+          null,
+          null,
+          null,
+          new InterpreterOutput(null));
+      InterpreterResult result = sparkInterpreter.interpret("spark", context);
       if (result.code() == InterpreterResult.Code.SUCCESS &&
           
result.message().get(0).getData().contains("org.apache.spark.sql.SparkSession"))
 {
         LOGGER.info("SparkSession is detected so we are using spark 2.x for 
session {}",
@@ -72,7 +90,7 @@ public class LivySparkSQLInterpreter extends 
BaseLivyInterpreter {
         isSpark2 = true;
       } else {
         // spark 1.x
-        result = sparkInterpreter.interpret("sqlContext", null, false, false);
+        result = sparkInterpreter.interpret("sqlContext", context);
         if (result.code() == InterpreterResult.Code.SUCCESS) {
           LOGGER.info("sqlContext is detected.");
         } else if (result.code() == InterpreterResult.Code.ERROR) {
@@ -81,7 +99,7 @@ public class LivySparkSQLInterpreter extends 
BaseLivyInterpreter {
           LOGGER.info("sqlContext is not detected, try to create SQLContext by 
ourselves");
           result = sparkInterpreter.interpret(
               "val sqlContext = new org.apache.spark.sql.SQLContext(sc)\n"
-                  + "import sqlContext.implicits._", null, false, false);
+                  + "import sqlContext.implicits._", context);
           if (result.code() == InterpreterResult.Code.ERROR) {
             throw new LivyException("Fail to create SQLContext," +
                 result.message().get(0).getData());
@@ -128,9 +146,7 @@ public class LivySparkSQLInterpreter extends 
BaseLivyInterpreter {
         sqlQuery = "sqlContext.sql(\"\"\"" + line + "\"\"\").show(" + 
maxResult + ", " +
             truncate + ")";
       }
-      InterpreterResult result = sparkInterpreter.interpret(sqlQuery, 
context.getParagraphId(),
-          this.displayAppInfo, true);
-
+      InterpreterResult result = sparkInterpreter.interpret(sqlQuery, context);
       if (result.code() == InterpreterResult.Code.SUCCESS) {
         InterpreterResult result2 = new 
InterpreterResult(InterpreterResult.Code.SUCCESS);
         for (InterpreterResultMessage message : result.message()) {
@@ -248,12 +264,16 @@ public class LivySparkSQLInterpreter extends 
BaseLivyInterpreter {
 
   @Override
   public void cancel(InterpreterContext context) {
-    sparkInterpreter.cancel(context);
+    if (this.sparkInterpreter != null) {
+      sparkInterpreter.cancel(context);
+    }
   }
 
   @Override
   public void close() {
-    this.sparkInterpreter.close();
+    if (this.sparkInterpreter != null) {
+      this.sparkInterpreter.close();
+    }
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/3745e65b/livy/src/main/java/org/apache/zeppelin/livy/LivyVersion.java
----------------------------------------------------------------------
diff --git a/livy/src/main/java/org/apache/zeppelin/livy/LivyVersion.java 
b/livy/src/main/java/org/apache/zeppelin/livy/LivyVersion.java
index 7cfecfb..81bb8d4 100644
--- a/livy/src/main/java/org/apache/zeppelin/livy/LivyVersion.java
+++ b/livy/src/main/java/org/apache/zeppelin/livy/LivyVersion.java
@@ -29,6 +29,7 @@ public class LivyVersion {
   protected static final LivyVersion LIVY_0_2_0 = 
LivyVersion.fromVersionString("0.2.0");
   protected static final LivyVersion LIVY_0_3_0 = 
LivyVersion.fromVersionString("0.3.0");
   protected static final LivyVersion LIVY_0_4_0 = 
LivyVersion.fromVersionString("0.4.0");
+  protected static final LivyVersion LIVY_0_5_0 = 
LivyVersion.fromVersionString("0.5.0");
 
   private int version;
   private String versionString;
@@ -79,6 +80,10 @@ public class LivyVersion {
     return this.newerThanEquals(LIVY_0_4_0);
   }
 
+  public boolean isSharedSupported() {
+    return this.newerThanEquals(LIVY_0_5_0);
+  }
+
   public boolean equals(Object versionToCompare) {
     return version == ((LivyVersion) versionToCompare).version;
   }

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/3745e65b/livy/src/main/resources/interpreter-setting.json
----------------------------------------------------------------------
diff --git a/livy/src/main/resources/interpreter-setting.json 
b/livy/src/main/resources/interpreter-setting.json
index 2d72487..cecacac 100644
--- a/livy/src/main/resources/interpreter-setting.json
+++ b/livy/src/main/resources/interpreter-setting.json
@@ -227,5 +227,21 @@
       "editOnDblClick": false,
       "completionKey": "TAB"
     }
+  },
+  {
+    "group": "livy",
+    "name": "shared",
+    "className": "org.apache.zeppelin.livy.LivySharedInterpreter",
+    "properties": {
+    },
+    "option": {
+      "remote": true,
+      "port": -1,
+      "perNote": "shared",
+      "perUser": "scoped",
+      "isExistingProcess": false,
+      "setPermission": false,
+      "users": []
+    }
   }
 ]

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/3745e65b/livy/src/test/java/org/apache/zeppelin/livy/LivyInterpreterIT.java
----------------------------------------------------------------------
diff --git a/livy/src/test/java/org/apache/zeppelin/livy/LivyInterpreterIT.java 
b/livy/src/test/java/org/apache/zeppelin/livy/LivyInterpreterIT.java
index ef3eabe..3dfeb36 100644
--- a/livy/src/test/java/org/apache/zeppelin/livy/LivyInterpreterIT.java
+++ b/livy/src/test/java/org/apache/zeppelin/livy/LivyInterpreterIT.java
@@ -34,6 +34,7 @@ import java.util.Properties;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
 
 public class LivyInterpreterIT {
 
@@ -76,7 +77,7 @@ public class LivyInterpreterIT {
   }
 
 
-//  @Test
+  @Test
   public void testSparkInterpreterRDD() throws InterpreterException {
     if (!checkPreCondition()) {
       return;
@@ -197,7 +198,7 @@ public class LivyInterpreterIT {
   }
 
 
-//  @Test
+  @Test
   public void testSparkInterpreterDataFrame() throws InterpreterException {
     if (!checkPreCondition()) {
       return;
@@ -285,7 +286,7 @@ public class LivyInterpreterIT {
     }
   }
 
-//  @Test
+  @Test
   public void testSparkSQLInterpreter() throws InterpreterException {
     if (!checkPreCondition()) {
       return;
@@ -320,7 +321,7 @@ public class LivyInterpreterIT {
   }
 
 
-//  @Test
+  @Test
   public void testSparkSQLCancellation() throws InterpreterException {
     if (!checkPreCondition()) {
       return;
@@ -401,7 +402,7 @@ public class LivyInterpreterIT {
     }
   }
 
-//  @Test
+  @Test
   public void testStringWithTruncation() throws InterpreterException {
     if (!checkPreCondition()) {
       return;
@@ -462,7 +463,7 @@ public class LivyInterpreterIT {
   }
 
 
-//  @Test
+  @Test
   public void testStringWithoutTruncation() throws InterpreterException {
     if (!checkPreCondition()) {
       return;
@@ -534,6 +535,7 @@ public class LivyInterpreterIT {
     }
 
     final LivyPySparkInterpreter pysparkInterpreter = new 
LivyPySparkInterpreter(properties);
+    pysparkInterpreter.setInterpreterGroup(mock(InterpreterGroup.class));
     AuthenticationInfo authInfo = new AuthenticationInfo("user1");
     MyInterpreterOutputListener outputListener = new 
MyInterpreterOutputListener();
     InterpreterOutput output = new InterpreterOutput(outputListener);
@@ -647,7 +649,7 @@ public class LivyInterpreterIT {
     }
   }
 
-//  @Test
+  @Test
   public void testSparkInterpreterWithDisplayAppInfo() throws 
InterpreterException {
     if (!checkPreCondition()) {
       return;
@@ -686,13 +688,15 @@ public class LivyInterpreterIT {
     }
   }
 
-//  @Test
+  @Test
   public void testSparkRInterpreter() throws LivyException, 
InterpreterException {
     if (!checkPreCondition()) {
       return;
     }
 
     final LivySparkRInterpreter sparkRInterpreter = new 
LivySparkRInterpreter(properties);
+    sparkRInterpreter.setInterpreterGroup(mock(InterpreterGroup.class));
+
     try {
       sparkRInterpreter.getLivyVersion();
     } catch (APINotFoundException e) {
@@ -749,8 +753,7 @@ public class LivyInterpreterIT {
 
       // error
       result = sparkRInterpreter.interpret("cat(a)", context);
-      //TODO @zjffdu, it should be ERROR, it is due to bug of LIVY-313
-      assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+      assertEquals(InterpreterResult.Code.ERROR, result.code());
       assertEquals(InterpreterResult.Type.TEXT, 
result.message().get(0).getType());
       assertTrue(result.message().get(0).getData().contains("object 'a' not 
found"));
     } finally {
@@ -758,7 +761,7 @@ public class LivyInterpreterIT {
     }
   }
 
-//  @Test
+  @Test
   public void testLivyTutorialNote() throws IOException, InterpreterException {
     if (!checkPreCondition()) {
       return;
@@ -796,6 +799,114 @@ public class LivyInterpreterIT {
     }
   }
 
+  @Test
+  public void testSharedInterpreter() throws InterpreterException {
+    if (!checkPreCondition()) {
+      return;
+    }
+    InterpreterGroup interpreterGroup = new InterpreterGroup("group_1");
+    interpreterGroup.put("session_1", new ArrayList<Interpreter>());
+    LazyOpenInterpreter sparkInterpreter = new LazyOpenInterpreter(
+        new LivySparkInterpreter(properties));
+    sparkInterpreter.setInterpreterGroup(interpreterGroup);
+    interpreterGroup.get("session_1").add(sparkInterpreter);
+
+    LazyOpenInterpreter sqlInterpreter = new LazyOpenInterpreter(
+        new LivySparkSQLInterpreter(properties));
+    interpreterGroup.get("session_1").add(sqlInterpreter);
+    sqlInterpreter.setInterpreterGroup(interpreterGroup);
+
+    LazyOpenInterpreter pysparkInterpreter = new LazyOpenInterpreter(
+        new LivyPySparkInterpreter(properties));
+    interpreterGroup.get("session_1").add(pysparkInterpreter);
+    pysparkInterpreter.setInterpreterGroup(interpreterGroup);
+
+    LazyOpenInterpreter sparkRInterpreter = new LazyOpenInterpreter(
+        new LivySparkRInterpreter(properties));
+    interpreterGroup.get("session_1").add(sparkRInterpreter);
+    sparkRInterpreter.setInterpreterGroup(interpreterGroup);
+
+    LazyOpenInterpreter sharedInterpreter = new LazyOpenInterpreter(
+        new LivySharedInterpreter(properties));
+    interpreterGroup.get("session_1").add(sharedInterpreter);
+    sharedInterpreter.setInterpreterGroup(interpreterGroup);
+
+    sparkInterpreter.open();
+    sqlInterpreter.open();
+    pysparkInterpreter.open();
+    sparkRInterpreter.open();
+
+    try {
+      AuthenticationInfo authInfo = new AuthenticationInfo("user1");
+      MyInterpreterOutputListener outputListener = new 
MyInterpreterOutputListener();
+      InterpreterOutput output = new InterpreterOutput(outputListener);
+      InterpreterContext context = new InterpreterContext("noteId", 
"paragraphId", "livy.sql",
+          "title", "text", authInfo, null, null, null, null, null, null, 
output);
+      // detect spark version
+      InterpreterResult result = sparkInterpreter.interpret("sc.version", 
context);
+      assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+      assertEquals(1, result.message().size());
+
+      boolean isSpark2 = 
isSpark2((BaseLivyInterpreter)sparkInterpreter.getInnerInterpreter(), context);
+
+      if (!isSpark2) {
+        result = sparkInterpreter.interpret(
+            "val 
df=sqlContext.createDataFrame(Seq((\"hello\",20))).toDF(\"col_1\", \"col_2\")\n"
+                + "df.collect()", context);
+        assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+        assertEquals(1, result.message().size());
+        assertTrue(result.message().get(0).getData()
+            .contains("Array[org.apache.spark.sql.Row] = Array([hello,20])"));
+        sparkInterpreter.interpret("df.registerTempTable(\"df\")", context);
+
+        // access table from pyspark
+        result = pysparkInterpreter.interpret("sqlContext.sql(\"select * from 
df\").show()", context);
+        assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+        assertEquals(1, result.message().size());
+        assertTrue(result.message().get(0).getData()
+            .contains("+-----+-----+\n" +
+                "|col_1|col_2|\n" +
+                "+-----+-----+\n" +
+                "|hello|   20|\n" +
+                "+-----+-----+"));
+
+        // access table from sparkr
+        result = sparkRInterpreter.interpret("head(sql(sqlContext, \"select * 
from df\"))", context);
+        assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+        assertEquals(1, result.message().size());
+        assertTrue(result.message().get(0).getData().contains("col_1 col_2\n1 
hello    20"));
+      } else {
+        result = sparkInterpreter.interpret(
+            "val df=spark.createDataFrame(Seq((\"hello\",20))).toDF(\"col_1\", 
\"col_2\")\n"
+                + "df.collect()", context);
+        assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+        assertEquals(1, result.message().size());
+        assertTrue(result.message().get(0).getData()
+            .contains("Array[org.apache.spark.sql.Row] = Array([hello,20])"));
+        sparkInterpreter.interpret("df.registerTempTable(\"df\")", context);
+
+        // access table from pyspark
+        result = pysparkInterpreter.interpret("spark.sql(\"select * from 
df\").show()", context);
+        assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+        assertEquals(1, result.message().size());
+        assertTrue(result.message().get(0).getData()
+            .contains("+-----+-----+\n" +
+                      "|col_1|col_2|\n" +
+                      "+-----+-----+\n" +
+                      "|hello|   20|\n" +
+                      "+-----+-----+"));
+
+        // access table from sparkr
+        result = sparkRInterpreter.interpret("head(sql(\"select * from 
df\"))", context);
+        assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+        assertEquals(1, result.message().size());
+        assertTrue(result.message().get(0).getData().contains("col_1 col_2\n1 
hello    20"));
+      }
+    } finally {
+      sparkInterpreter.close();
+      sqlInterpreter.close();
+    }
+  }
 
   private boolean isSpark2(BaseLivyInterpreter interpreter, InterpreterContext 
context) {
     InterpreterResult result = null;

Reply via email to