You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zeppelin.apache.org by zj...@apache.org on 2018/04/16 06:18:04 UTC
[3/3] zeppelin git commit: ZEPPELIN-3375: Make PySparkInterpreter
extends PythonInterpreter
ZEPPELIN-3375: Make PySparkInterpreter extends PythonInterpreter
### What is this PR for?
This PR is trying to remove the code duplication between PySparkInterpreter and PythonInterpreter. So here's the main things this PR did:
* PySparkInterpreter extends PythonInterpreter
* PySparkInterpreterTest extends PythonInterpreterTest so that we can verify PySparkInterpreter can do whatever PythonInterpreter can do
* Move interpreter/lib/python/backend_zinline.py and interpreter/lib/python/mpl_config.py into python module, so that python module can ship these resources together.
### What type of PR is it?
[ Improvement | Refactoring]
### Todos
* [ ] - Task
### What is the Jira issue?
* https://issues.apache.org/jira/browse/ZEPPELIN-3375
### How should this be tested?
* CI pass
### 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 <zj...@apache.org>
Closes #2919 from zjffdu/ZEPPELIN-3375 and squashes the following commits:
738c6c5 [Jeff Zhang] ZEPPELIN-3375. Make PySparkInterpreter extends PythonInterpreter
Project: http://git-wip-us.apache.org/repos/asf/zeppelin/repo
Commit: http://git-wip-us.apache.org/repos/asf/zeppelin/commit/0a97446a
Tree: http://git-wip-us.apache.org/repos/asf/zeppelin/tree/0a97446a
Diff: http://git-wip-us.apache.org/repos/asf/zeppelin/diff/0a97446a
Branch: refs/heads/master
Commit: 0a97446a70f6294a3efb071bb9a70601f885840b
Parents: 7aa94ce
Author: Jeff Zhang <zj...@apache.org>
Authored: Sun Apr 1 21:26:28 2018 +0800
Committer: Jeff Zhang <zj...@apache.org>
Committed: Mon Apr 16 14:17:54 2018 +0800
----------------------------------------------------------------------
interpreter/lib/python/backend_zinline.py | 317 --------
interpreter/lib/python/mpl_config.py | 105 ---
python/pom.xml | 25 -
.../zeppelin/python/IPythonInterpreter.java | 25 +-
.../zeppelin/python/PythonCondaInterpreter.java | 77 +-
.../python/PythonDockerInterpreter.java | 19 +-
.../zeppelin/python/PythonInterpreter.java | 760 ++++++++++---------
.../python/PythonInterpreterPandasSql.java | 2 +-
.../main/resources/python/backend_zinline.py | 317 ++++++++
python/src/main/resources/python/mpl_config.py | 105 +++
.../main/resources/python/py4j-src-0.9.2.zip | Bin 0 -> 71355 bytes
.../main/resources/python/zeppelin_python.py | 159 ++--
.../python/BasePythonInterpreterTest.java | 331 ++++++++
.../zeppelin/python/IPythonInterpreterTest.java | 383 ++--------
.../python/PythonCondaInterpreterTest.java | 6 +-
.../python/PythonDockerInterpreterTest.java | 23 +-
.../zeppelin/python/PythonInterpreterTest.java | 165 ++--
python/src/test/resources/log4j.properties | 17 +-
spark/interpreter/pom.xml | 4 +-
.../zeppelin/spark/IPySparkInterpreter.java | 16 +-
.../zeppelin/spark/NewSparkInterpreter.java | 7 +-
.../zeppelin/spark/PySparkInterpreter.java | 581 ++------------
.../main/resources/python/zeppelin_pyspark.py | 215 +-----
.../zeppelin/spark/IPySparkInterpreterTest.java | 87 ++-
.../zeppelin/spark/OldSparkInterpreterTest.java | 4 +-
.../zeppelin/spark/PySparkInterpreterTest.java | 169 ++---
.../zeppelin/spark/SparkRInterpreterTest.java | 16 +-
.../src/test/resources/log4j.properties | 6 +-
.../zeppelin/interpreter/InterpreterGroup.java | 13 +
.../zeppelin/rest/ZeppelinSparkClusterTest.java | 17 +-
.../interpreter/InterpreterSetting.java | 3 +-
31 files changed, 1678 insertions(+), 2296 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/0a97446a/interpreter/lib/python/backend_zinline.py
----------------------------------------------------------------------
diff --git a/interpreter/lib/python/backend_zinline.py b/interpreter/lib/python/backend_zinline.py
deleted file mode 100644
index 1c84699..0000000
--- a/interpreter/lib/python/backend_zinline.py
+++ /dev/null
@@ -1,317 +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.
-
-# This file provides a static (non-interactive) matplotlib plotting backend
-# for zeppelin notebooks for use with the python/pyspark interpreters
-
-from __future__ import print_function
-
-import sys
-import uuid
-import warnings
-import base64
-from io import BytesIO
-try:
- from StringIO import StringIO
-except ImportError:
- from io import StringIO
-
-import mpl_config
-import matplotlib
-from matplotlib._pylab_helpers import Gcf
-from matplotlib.backends.backend_agg import new_figure_manager, FigureCanvasAgg
-from matplotlib.backend_bases import ShowBase, FigureManagerBase
-from matplotlib.figure import Figure
-
-########################################################################
-#
-# The following functions and classes are for pylab and implement
-# window/figure managers, etc...
-#
-########################################################################
-
-class Show(ShowBase):
- """
- A callable object that displays the figures to the screen. Valid kwargs
- include figure width and height (in units supported by the div tag), block
- (allows users to override blocking behavior regardless of whether or not
- interactive mode is enabled, currently unused) and close (Implicitly call
- matplotlib.pyplot.close('all') with each call to show()).
- """
- def __call__(self, close=None, block=None, **kwargs):
- if close is None:
- close = mpl_config.get('close')
- try:
- managers = Gcf.get_all_fig_managers()
- if not managers:
- return
-
- # Tell zeppelin that the output will be html using the %html magic
- # We want to do this only once to avoid seeing "%html" printed
- # directly to the outout when multiple figures are displayed from
- # one paragraph.
- if mpl_config.get('angular'):
- print('%angular')
- else:
- print('%html')
-
- # Show all open figures
- for manager in managers:
- manager.show(**kwargs)
- finally:
- # This closes all the figures if close is set to True.
- if close and Gcf.get_all_fig_managers():
- Gcf.destroy_all()
-
-
-class FigureCanvasZInline(FigureCanvasAgg):
- """
- The canvas the figure renders into. Calls the draw and print fig
- methods, creates the renderers, etc...
- """
- def get_bytes(self, **kwargs):
- """
- Get the byte representation of the figure.
- Should only be used with jpg/png formats.
- """
- # Make sure format is correct
- fmt = kwargs.get('format', mpl_config.get('format'))
- if fmt == 'svg':
- raise ValueError("get_bytes() does not support svg, use png or jpg")
-
- # Express the image as bytes
- buf = BytesIO()
- self.print_figure(buf, **kwargs)
- fmt = fmt.encode()
- if sys.version_info >= (3, 4) and sys.version_info < (3, 5):
- byte_str = bytes("data:image/%s;base64," %fmt, "utf-8")
- else:
- byte_str = b"data:image/%s;base64," %fmt
- byte_str += base64.b64encode(buf.getvalue())
-
- # Python3 forces all strings to default to unicode, but for raster image
- # formats (eg png, jpg), we want to work with bytes. Thus this step is
- # needed to ensure compatability for all python versions.
- byte_str = byte_str.decode('ascii')
- buf.close()
- return byte_str
-
- def get_svg(self, **kwargs):
- """
- Get the svg representation of the figure.
- Should only be used with svg format.
- """
- # Make sure format is correct
- fmt = kwargs.get('format', mpl_config.get('format'))
- if fmt != 'svg':
- raise ValueError("get_svg() does not support png or jpg, use svg")
-
- # For SVG the data string has to be unicode, not bytes
- buf = StringIO()
- self.print_figure(buf, **kwargs)
- svg_str = buf.getvalue()
- buf.close()
- return svg_str
-
- def draw_idle(self, *args, **kwargs):
- """
- Called when the figure gets updated (eg through a plotting command).
- This is overriden to allow open figures to be reshown after they
- are updated when mpl_config.get('close') is False.
- """
- if not self._is_idle_drawing:
- with self._idle_draw_cntx():
- self.draw(*args, **kwargs)
- draw_if_interactive()
-
-
-class FigureManagerZInline(FigureManagerBase):
- """
- Wrap everything up into a window for the pylab interface
- """
- def __init__(self, canvas, num):
- FigureManagerBase.__init__(self, canvas, num)
- self.fig_id = "figure_{0}".format(uuid.uuid4().hex)
- self._shown = False
-
- def angular_bind(self, **kwargs):
- """
- Bind figure data to Zeppelin's Angular Object Registry.
- If mpl_config("angular") is True and PY4J is supported, this allows
- for the possibility to interactively update a figure from a separate
- paragraph without having to display it multiple times.
- """
- # This doesn't work for SVG so make sure it's not our format
- fmt = kwargs.get('format', mpl_config.get('format'))
- if fmt == 'svg':
- return
-
- # Get the figure data as a byte array
- src = self.canvas.get_bytes(**kwargs)
-
- # Flag to determine whether or not to use
- # zeppelin's angular display system
- angular = mpl_config.get('angular')
-
- # ZeppelinContext instance (requires PY4J)
- context = mpl_config.get('context')
-
- # Finally we must ensure that automatic closing is set to False,
- # as otherwise using the angular display system is pointless
- close = mpl_config.get('close')
-
- # If above conditions are met, bind the figure data to
- # the Angular Object Registry.
- if not close and angular:
- if hasattr(context, 'angularBind'):
- # Binding is performed through figure ID to ensure this works
- # if multiple figures are open
- context.angularBind(self.fig_id, src)
-
- # Zeppelin will automatically replace this value even if it
- # is updated from another pargraph thanks to the {{}} notation
- src = "{{%s}}" %self.fig_id
- else:
- warnings.warn("Cannot bind figure to Angular Object Registry. "
- "Check if PY4J is installed.")
- return src
-
- def angular_unbind(self):
- """
- Unbind figure from angular display system.
- """
- context = mpl_config.get('context')
- if hasattr(context, 'angularUnbind'):
- context.angularUnbind(self.fig_id)
-
- def destroy(self):
- """
- Called when close=True or implicitly by pyplot.close().
- Overriden to automatically clean up the angular object registry.
- """
- self.angular_unbind()
-
- def show(self, **kwargs):
- if not self._shown:
- zdisplay(self.canvas.figure, **kwargs)
- else:
- self.canvas.draw_idle()
- self.angular_bind(**kwargs)
-
- self._shown = True
-
-
-def draw_if_interactive():
- """
- If interactive mode is on, this allows for updating properties of
- the figure when each new plotting command is called.
- """
- manager = Gcf.get_active()
- interactive = matplotlib.is_interactive()
- angular = mpl_config.get('angular')
-
- # Don't bother continuing if we aren't in interactive mode
- # or if there are no active figures. Also pointless to continue
- # in angular mode as we don't want to reshow the figure.
- if not interactive or angular or manager is None:
- return
-
- # Allow for figure to be reshown if close is false since
- # this function call implies that it has been updated
- if not mpl_config.get('close'):
- manager._shown = False
-
-
-def new_figure_manager(num, *args, **kwargs):
- """
- Create a new figure manager instance
- """
- # if a main-level app must be created, this (and
- # new_figure_manager_given_figure) is the usual place to
- # do it -- see backend_wx, backend_wxagg and backend_tkagg for
- # examples. Not all GUIs require explicit instantiation of a
- # main-level app (egg backend_gtk, backend_gtkagg) for pylab
- FigureClass = kwargs.pop('FigureClass', Figure)
- thisFig = FigureClass(*args, **kwargs)
- return new_figure_manager_given_figure(num, thisFig)
-
-
-def new_figure_manager_given_figure(num, figure):
- """
- Create a new figure manager instance for the given figure.
- """
- canvas = FigureCanvasZInline(figure)
- manager = FigureManagerZInline(canvas, num)
- return manager
-
-
-########################################################################
-#
-# Backend specific functions
-#
-########################################################################
-
-def zdisplay(fig, **kwargs):
- """
- Publishes a matplotlib figure to the notebook paragraph output.
- """
- # kwargs can be width or height (in units supported by div tag)
- width = kwargs.pop('width', 'auto')
- height = kwargs.pop('height', 'auto')
- fmt = kwargs.get('format', mpl_config.get('format'))
-
- # Check if format is supported
- supported_formats = mpl_config.get('supported_formats')
- if fmt not in supported_formats:
- raise ValueError("Unsupported format %s" %fmt)
-
- # For SVG the data string has to be unicode, not bytes
- if fmt == 'svg':
- img = fig.canvas.get_svg(**kwargs)
-
- # This is needed to ensure the SVG image is the correct size.
- # We should find a better way to do this...
- width = '{}px'.format(mpl_config.get('width'))
- height = '{}px'.format(mpl_config.get('height'))
- else:
- # Express the image as bytes
- src = fig.canvas.manager.angular_bind(**kwargs)
- img = "<img src={src} style='width={width};height:{height}'>"
- img = img.format(src=src, width=width, height=height)
-
- # Print the image to the notebook paragraph via the %html magic
- html = "<div style='width:{width};height:{height}'>{img}<div>"
- print(html.format(width=width, height=height, img=img))
-
-def displayhook():
- """
- Called post paragraph execution if interactive mode is on
- """
- if matplotlib.is_interactive():
- show()
-
-########################################################################
-#
-# Now just provide the standard names that backend.__init__ is expecting
-#
-########################################################################
-
-# Create a reference to the show function we are using. This is what actually
-# gets called by matplotlib.pyplot.show().
-show = Show()
-
-# Default FigureCanvas and FigureManager classes to use from the backend
-FigureCanvas = FigureCanvasZInline
-FigureManager = FigureManagerZInline
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/0a97446a/interpreter/lib/python/mpl_config.py
----------------------------------------------------------------------
diff --git a/interpreter/lib/python/mpl_config.py b/interpreter/lib/python/mpl_config.py
deleted file mode 100644
index 5c60893..0000000
--- a/interpreter/lib/python/mpl_config.py
+++ /dev/null
@@ -1,105 +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.
-
-# This module provides utitlites for users to configure the inline plotting
-# backend through a PyZeppelinContext instance (eg, through z.configure_mpl())
-
-import matplotlib
-
-def configure(**kwargs):
- """
- Generic configure function.
- Usage: configure(prop1='foo', prop2='bar', ...)
- Currently supported zeppelin-specific properties are:
- interactive - If true show all figures without explicit call to show()
- via a post-execute hook.
- angular - If true, bind figures to angular display system.
- close - If true, close all figures once shown.
- width, height - Default width / height of the figure in pixels.
- fontsize - Font size.
- dpi - dpi of the figure.
- fmt - Figure format
- supported_formats - Supported Figure formats ()
- context - ZeppelinContext instance (requires PY4J)
- """
- _config.update(**kwargs)
-
- # Broadcast relevant changes to matplotlib RC
- _on_config_change()
-
-
-def get(key):
- """
- Get the configuration info given a key
- """
- return _config[key]
-
-
-def _on_config_change():
- # dpi
- dpi = _config['dpi']
-
- # For older versions of matplotlib, savefig.dpi is not synced with
- # figure.dpi by default
- matplotlib.rcParams['figure.dpi'] = dpi
- if matplotlib.__version__ < '2.0.0':
- matplotlib.rcParams['savefig.dpi'] = dpi
-
- # Width and height
- width = float(_config['width']) / dpi
- height = float(_config['height']) / dpi
- matplotlib.rcParams['figure.figsize'] = (width, height)
-
- # Font size
- fontsize = _config['fontsize']
- matplotlib.rcParams['font.size'] = fontsize
-
- # Default Figure Format
- fmt = _config['format']
- supported_formats = _config['supported_formats']
- if fmt not in supported_formats:
- raise ValueError("Unsupported format %s" %fmt)
-
- if matplotlib.__version__ < '1.2.0':
- matplotlib.rcParams.update({'savefig.format': fmt})
- else:
- matplotlib.rcParams['savefig.format'] = fmt
-
- # Interactive mode
- interactive = _config['interactive']
- matplotlib.interactive(interactive)
-
-
-def _init_config():
- dpi = matplotlib.rcParams['figure.dpi']
- if matplotlib.__version__ < '1.2.0':
- matplotlib.rcParams.update({'savefig.format': 'png'})
- fmt = matplotlib.rcParams['savefig.format']
- width, height = matplotlib.rcParams['figure.figsize']
- fontsize = matplotlib.rcParams['font.size']
- _config['dpi'] = dpi
- _config['format'] = fmt
- _config['width'] = width*dpi
- _config['height'] = height*dpi
- _config['fontsize'] = fontsize
- _config['close'] = True
- _config['interactive'] = matplotlib.is_interactive()
- _config['angular'] = False
- _config['supported_formats'] = ['png', 'jpg', 'svg']
- _config['context'] = None
-
-
-_config = {}
-_init_config()
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/0a97446a/python/pom.xml
----------------------------------------------------------------------
diff --git a/python/pom.xml b/python/pom.xml
index c14d4b1..289755d 100644
--- a/python/pom.xml
+++ b/python/pom.xml
@@ -40,8 +40,6 @@
**/PythonInterpreterPandasSqlTest.java,
**/PythonInterpreterMatplotlibTest.java
</python.test.exclude>
- <pypi.repo.url>https://pypi.python.org/packages</pypi.repo.url>
- <python.py4j.repo.folder>/64/5c/01e13b68e8caafece40d549f232c9b5677ad1016071a48d04cc3895acaa3</python.py4j.repo.folder>
<grpc.version>1.4.0</grpc.version>
<plugin.shade.version>2.4.1</plugin.shade.version>
</properties>
@@ -138,34 +136,11 @@
</plugin>
<plugin>
- <groupId>org.codehaus.mojo</groupId>
- <artifactId>wagon-maven-plugin</artifactId>
- <version>1.0</version>
- <executions>
- <execution>
- <phase>package</phase>
- <goals><goal>download-single</goal></goals>
- <configuration>
- <url>${pypi.repo.url}${python.py4j.repo.folder}</url>
- <fromFile>py4j-${python.py4j.version}.zip</fromFile>
- <toFile>${project.build.directory}/../../interpreter/python/py4j-${python.py4j.version}.zip</toFile>
- </configuration>
- </execution>
- </executions>
- </plugin>
-
- <plugin>
<artifactId>maven-antrun-plugin</artifactId>
<version>1.7</version>
<executions>
<execution>
<phase>package</phase>
- <configuration>
- <target>
- <unzip src="${project.build.directory}/../../interpreter/python/py4j-${python.py4j.version}.zip"
- dest="${project.build.directory}/../../interpreter/python"/>
- </target>
- </configuration>
<goals>
<goal>run</goal>
</goals>
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/0a97446a/python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java
----------------------------------------------------------------------
diff --git a/python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java b/python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java
index 4fe50ee..2daa986 100644
--- a/python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java
+++ b/python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java
@@ -243,16 +243,21 @@ public class IPythonInterpreter extends Interpreter implements ExecuteResultHand
private void launchIPythonKernel(int ipythonPort)
throws IOException, URISyntaxException {
// copy the python scripts to a temp directory, then launch ipython kernel in that folder
- File tmpPythonScriptFolder = Files.createTempDirectory("zeppelin_ipython").toFile();
+ File pythonWorkDir = Files.createTempDirectory("zeppelin_ipython").toFile();
String[] ipythonScripts = {"ipython_server.py", "ipython_pb2.py", "ipython_pb2_grpc.py"};
for (String ipythonScript : ipythonScripts) {
URL url = getClass().getClassLoader().getResource("grpc/python"
+ "/" + ipythonScript);
- FileUtils.copyURLToFile(url, new File(tmpPythonScriptFolder, ipythonScript));
+ FileUtils.copyURLToFile(url, new File(pythonWorkDir, ipythonScript));
}
+ //TODO(zjffdu) don't do hard code on py4j here
+ File py4jDestFile = new File(pythonWorkDir, "py4j-src-0.9.2.zip");
+ FileUtils.copyURLToFile(getClass().getClassLoader().getResource(
+ "python/py4j-src-0.9.2.zip"), py4jDestFile);
+
CommandLine cmd = CommandLine.parse(pythonExecutable);
- cmd.addArgument(tmpPythonScriptFolder.getAbsolutePath() + "/ipython_server.py");
+ cmd.addArgument(pythonWorkDir.getAbsolutePath() + "/ipython_server.py");
cmd.addArgument(ipythonPort + "");
DefaultExecutor executor = new DefaultExecutor();
ProcessLogOutputStream processOutput = new ProcessLogOutputStream(LOGGER);
@@ -261,20 +266,12 @@ public class IPythonInterpreter extends Interpreter implements ExecuteResultHand
executor.setWatchdog(watchDog);
if (useBuiltinPy4j) {
- String py4jLibPath = null;
- if (System.getenv("ZEPPELIN_HOME") != null) {
- py4jLibPath = System.getenv("ZEPPELIN_HOME") + File.separator
- + PythonInterpreter.ZEPPELIN_PY4JPATH;
- } else {
- Path workingPath = Paths.get("..").toAbsolutePath();
- py4jLibPath = workingPath + File.separator + PythonInterpreter.ZEPPELIN_PY4JPATH;
- }
if (additionalPythonPath != null) {
// put the py4j at the end, because additionalPythonPath may already contain py4j.
// e.g. PySparkInterpreter
- additionalPythonPath = additionalPythonPath + ":" + py4jLibPath;
+ additionalPythonPath = additionalPythonPath + ":" + py4jDestFile.getAbsolutePath();
} else {
- additionalPythonPath = py4jLibPath;
+ additionalPythonPath = py4jDestFile.getAbsolutePath();
}
}
@@ -326,7 +323,7 @@ public class IPythonInterpreter extends Interpreter implements ExecuteResultHand
@Override
public void close() throws InterpreterException {
if (watchDog != null) {
- LOGGER.debug("Kill IPython Process");
+ LOGGER.info("Kill IPython Process");
ipythonClient.stop(StopRequest.newBuilder().build());
watchDog.destroyProcess();
gatewayServer.shutdown();
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/0a97446a/python/src/main/java/org/apache/zeppelin/python/PythonCondaInterpreter.java
----------------------------------------------------------------------
diff --git a/python/src/main/java/org/apache/zeppelin/python/PythonCondaInterpreter.java b/python/src/main/java/org/apache/zeppelin/python/PythonCondaInterpreter.java
index 887beb8..8d3e972 100644
--- a/python/src/main/java/org/apache/zeppelin/python/PythonCondaInterpreter.java
+++ b/python/src/main/java/org/apache/zeppelin/python/PythonCondaInterpreter.java
@@ -31,9 +31,10 @@ import java.util.regex.Pattern;
/**
* Conda support
+ * TODO(zjffdu) Add removing conda env
*/
public class PythonCondaInterpreter extends Interpreter {
- Logger logger = LoggerFactory.getLogger(PythonCondaInterpreter.class);
+ private static Logger logger = LoggerFactory.getLogger(PythonCondaInterpreter.class);
public static final String ZEPPELIN_PYTHON = "zeppelin.python";
public static final String CONDA_PYTHON_PATH = "/bin/python";
public static final String DEFAULT_ZEPPELIN_PYTHON = "python";
@@ -145,33 +146,22 @@ public class PythonCondaInterpreter extends Interpreter {
}
}
setCurrentCondaEnvName(envName);
- python.setPythonCommand(binPath);
+ python.setPythonExec(binPath);
}
private void restartPythonProcess() throws InterpreterException {
- PythonInterpreter python = getPythonInterpreter();
+ logger.debug("Restarting PythonInterpreter");
+ Interpreter python =
+ getInterpreterInTheSameSessionByClassName(PythonInterpreter.class.getName());
python.close();
python.open();
}
protected PythonInterpreter getPythonInterpreter() throws InterpreterException {
- LazyOpenInterpreter lazy = null;
PythonInterpreter python = null;
Interpreter p =
getInterpreterInTheSameSessionByClassName(PythonInterpreter.class.getName());
-
- while (p instanceof WrappedInterpreter) {
- if (p instanceof LazyOpenInterpreter) {
- lazy = (LazyOpenInterpreter) p;
- }
- p = ((WrappedInterpreter) p).getInnerInterpreter();
- }
- python = (PythonInterpreter) p;
-
- if (lazy != null) {
- lazy.open();
- }
- return python;
+ return (PythonInterpreter) ((LazyOpenInterpreter)p).getInnerInterpreter();
}
public static String runCondaCommandForTextOutput(String title, List<String> commands)
@@ -392,27 +382,50 @@ public class PythonCondaInterpreter extends Interpreter {
public static String runCommand(List<String> commands)
throws IOException, InterruptedException {
+ logger.info("Starting shell commands: " + StringUtils.join(commands, " "));
+ Process process = Runtime.getRuntime().exec(commands.toArray(new String[0]));
+ StreamGobbler errorGobbler = new StreamGobbler(process.getErrorStream());
+ StreamGobbler outputGobbler = new StreamGobbler(process.getInputStream());
+ errorGobbler.start();
+ outputGobbler.start();
+ if (process.waitFor() != 0) {
+ throw new IOException("Fail to run shell commands: " + StringUtils.join(commands, " "));
+ }
+ logger.info("Complete shell commands: " + StringUtils.join(commands, " "));
+ return outputGobbler.getOutput();
+ }
- StringBuilder sb = new StringBuilder();
+ private static class StreamGobbler extends Thread {
+ InputStream is;
+ StringBuilder output = new StringBuilder();
- ProcessBuilder builder = new ProcessBuilder(commands);
- builder.redirectErrorStream(true);
- Process process = builder.start();
- InputStream stdout = process.getInputStream();
- BufferedReader br = new BufferedReader(new InputStreamReader(stdout));
- String line;
- while ((line = br.readLine()) != null) {
- sb.append(line);
- sb.append("\n");
+ // reads everything from is until empty.
+ StreamGobbler(InputStream is) {
+ this.is = is;
}
- int r = process.waitFor(); // Let the process finish.
- if (r != 0) {
- throw new RuntimeException("Failed to execute `" +
- StringUtils.join(commands, " ") + "` exited with " + r);
+ public void run() {
+ try {
+ InputStreamReader isr = new InputStreamReader(is);
+ BufferedReader br = new BufferedReader(isr);
+ String line = null;
+ long startTime = System.currentTimeMillis();
+ while ( (line = br.readLine()) != null) {
+ output.append(line + "\n");
+ // logging per 5 seconds
+ if ((System.currentTimeMillis() - startTime) > 5000) {
+ logger.info(line);
+ startTime = System.currentTimeMillis();
+ }
+ }
+ } catch (IOException ioe) {
+ ioe.printStackTrace();
+ }
}
- return sb.toString();
+ public String getOutput() {
+ return output.toString();
+ }
}
public static String runCommand(String ... command)
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/0a97446a/python/src/main/java/org/apache/zeppelin/python/PythonDockerInterpreter.java
----------------------------------------------------------------------
diff --git a/python/src/main/java/org/apache/zeppelin/python/PythonDockerInterpreter.java b/python/src/main/java/org/apache/zeppelin/python/PythonDockerInterpreter.java
index 22f6c2e..b528efa 100644
--- a/python/src/main/java/org/apache/zeppelin/python/PythonDockerInterpreter.java
+++ b/python/src/main/java/org/apache/zeppelin/python/PythonDockerInterpreter.java
@@ -58,7 +58,7 @@ public class PythonDockerInterpreter extends Interpreter {
@Override
public InterpreterResult interpret(String st, InterpreterContext context)
throws InterpreterException {
- File pythonScript = new File(getPythonInterpreter().getScriptPath());
+ File pythonWorkDir = getPythonInterpreter().getPythonWorkDir();
InterpreterOutput out = context.out;
Matcher activateMatcher = activatePattern.matcher(st);
@@ -73,26 +73,23 @@ public class PythonDockerInterpreter extends Interpreter {
pull(out, image);
// mount pythonscript dir
- String mountPythonScript = "-v " +
- pythonScript.getParentFile().getAbsolutePath() +
- ":/_zeppelin_tmp ";
+ String mountPythonScript = "-v " + pythonWorkDir.getAbsolutePath() +
+ ":/_python_workdir ";
// mount zeppelin dir
- String mountPy4j = "-v " +
- zeppelinHome.getAbsolutePath() +
+ String mountPy4j = "-v " + zeppelinHome.getAbsolutePath() +
":/_zeppelin ";
// set PYTHONPATH
- String pythonPath = ":/_zeppelin/" + PythonInterpreter.ZEPPELIN_PY4JPATH + ":" +
- ":/_zeppelin/" + PythonInterpreter.ZEPPELIN_PYTHON_LIBS;
+ String pythonPath = ".:/_python_workdir/py4j-src-0.9.2.zip:/_python_workdir";
setPythonCommand("docker run -i --rm " +
mountPythonScript +
mountPy4j +
"-e PYTHONPATH=\"" + pythonPath + "\" " +
image + " " +
- getPythonInterpreter().getPythonBindPath() + " " +
- "/_zeppelin_tmp/" + pythonScript.getName());
+ getPythonInterpreter().getPythonExec() + " " +
+ "/_python_workdir/zeppelin_python.py");
restartPythonProcess();
out.clear();
return new InterpreterResult(InterpreterResult.Code.SUCCESS, "\"" + image + "\" activated");
@@ -108,7 +105,7 @@ public class PythonDockerInterpreter extends Interpreter {
public void setPythonCommand(String cmd) throws InterpreterException {
PythonInterpreter python = getPythonInterpreter();
- python.setPythonCommand(cmd);
+ python.setPythonExec(cmd);
}
private void printUsage(InterpreterOutput out) {
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/0a97446a/python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java
----------------------------------------------------------------------
diff --git a/python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java b/python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java
index 178f79a..95cfc82 100644
--- a/python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java
+++ b/python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java
@@ -1,41 +1,24 @@
/*
-* 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.
-*/
+ * 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.python;
-import java.io.BufferedReader;
-import java.io.BufferedWriter;
-import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.OutputStreamWriter;
-import java.io.PipedInputStream;
-import java.io.PipedOutputStream;
-import java.net.*;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.regex.Pattern;
-
+import com.google.common.io.Files;
+import com.google.gson.Gson;
import org.apache.commons.exec.CommandLine;
import org.apache.commons.exec.DefaultExecutor;
import org.apache.commons.exec.ExecuteException;
@@ -45,239 +28,233 @@ import org.apache.commons.exec.PumpStreamHandler;
import org.apache.commons.exec.environment.EnvironmentUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
-import org.apache.zeppelin.display.GUI;
-import org.apache.zeppelin.interpreter.*;
-import org.apache.zeppelin.interpreter.InterpreterResult.Code;
+import org.apache.zeppelin.interpreter.BaseZeppelinContext;
+import org.apache.zeppelin.interpreter.Interpreter;
+import org.apache.zeppelin.interpreter.InterpreterContext;
+import org.apache.zeppelin.interpreter.InterpreterException;
+import org.apache.zeppelin.interpreter.InterpreterGroup;
import org.apache.zeppelin.interpreter.InterpreterHookRegistry.HookType;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.apache.zeppelin.interpreter.InterpreterResult.Code;
+import org.apache.zeppelin.interpreter.InterpreterResultMessage;
+import org.apache.zeppelin.interpreter.InvalidHookException;
+import org.apache.zeppelin.interpreter.LazyOpenInterpreter;
+import org.apache.zeppelin.interpreter.WrappedInterpreter;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterUtils;
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
import org.apache.zeppelin.interpreter.util.InterpreterOutputStream;
-import org.apache.zeppelin.scheduler.Job;
-import org.apache.zeppelin.scheduler.Scheduler;
-import org.apache.zeppelin.scheduler.SchedulerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
import py4j.GatewayServer;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.Inet4Address;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+
/**
- * Python interpreter for Zeppelin.
+ * Interpreter for Python, it is the first implementation of interpreter for Python, so with less
+ * features compared to IPythonInterpreter, but requires less prerequisites than
+ * IPythonInterpreter, only python installation is required.
*/
public class PythonInterpreter extends Interpreter implements ExecuteResultHandler {
- private static final Logger LOG = LoggerFactory.getLogger(PythonInterpreter.class);
- public static final String ZEPPELIN_PYTHON = "python/zeppelin_python.py";
- public static final String ZEPPELIN_CONTEXT = "python/zeppelin_context.py";
- public static final String ZEPPELIN_PY4JPATH = "interpreter/python/py4j-0.9.2/src";
- public static final String ZEPPELIN_PYTHON_LIBS = "interpreter/lib/python";
- public static final String DEFAULT_ZEPPELIN_PYTHON = "python";
- public static final String MAX_RESULT = "zeppelin.python.maxResult";
-
- private PythonZeppelinContext zeppelinContext;
- private InterpreterContext context;
- private Pattern errorInLastLine = Pattern.compile(".*(Error|Exception): .*$");
- private String pythonPath;
- private int maxResult;
- private String py4jLibPath;
- private String pythonLibPath;
-
- private String pythonCommand;
+ private static final Logger LOGGER = LoggerFactory.getLogger(PythonInterpreter.class);
+ private static final int MAX_TIMEOUT_SEC = 10;
private GatewayServer gatewayServer;
private DefaultExecutor executor;
- private int port;
- private InterpreterOutputStream outputStream;
- private BufferedWriter ins;
- private PipedInputStream in;
- private ByteArrayOutputStream input;
- private String scriptPath;
- boolean pythonscriptRunning = false;
- private static final int MAX_TIMEOUT_SEC = 10;
+ private File pythonWorkDir;
+ protected boolean useBuiltinPy4j = true;
- private long pythonPid = 0;
+ // used to forward output from python process to InterpreterOutput
+ private InterpreterOutputStream outputStream;
+ private AtomicBoolean pythonScriptRunning = new AtomicBoolean(false);
+ private AtomicBoolean pythonScriptInitialized = new AtomicBoolean(false);
+ private long pythonPid = -1;
private IPythonInterpreter iPythonInterpreter;
-
- Integer statementSetNotifier = new Integer(0);
+ private BaseZeppelinContext zeppelinContext;
+ private String condaPythonExec; // set by PythonCondaInterpreter
public PythonInterpreter(Properties property) {
super(property);
- try {
- File scriptFile = File.createTempFile("zeppelin_python-", ".py", new File("/tmp"));
- scriptPath = scriptFile.getAbsolutePath();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- private String workingDir() {
- URL myURL = getClass().getProtectionDomain().getCodeSource().getLocation();
- java.net.URI myURI = null;
- try {
- myURI = myURL.toURI();
- } catch (URISyntaxException e1)
- {}
- String path = java.nio.file.Paths.get(myURI).toFile().toString();
- return path;
}
- private void createPythonScript() throws InterpreterException {
- File out = new File(scriptPath);
-
- if (out.exists() && out.isDirectory()) {
- throw new InterpreterException("Can't create python script " + out.getAbsolutePath());
+ @Override
+ public void open() throws InterpreterException {
+ // try IPythonInterpreter first
+ iPythonInterpreter = getIPythonInterpreter();
+ if (getProperty("zeppelin.python.useIPython", "true").equals("true") &&
+ StringUtils.isEmpty(
+ iPythonInterpreter.checkIPythonPrerequisite(getPythonExec()))) {
+ try {
+ iPythonInterpreter.open();
+ LOGGER.info("IPython is available, Use IPythonInterpreter to replace PythonInterpreter");
+ return;
+ } catch (Exception e) {
+ iPythonInterpreter = null;
+ LOGGER.warn("Fail to open IPythonInterpreter", e);
+ }
}
- copyFile(out, ZEPPELIN_PYTHON);
- // copy zeppelin_context.py as well
- File zOut = new File(out.getParent() + "/zeppelin_context.py");
- copyFile(zOut, ZEPPELIN_CONTEXT);
-
- logger.info("File {} , {} created", scriptPath, zOut.getAbsolutePath());
- }
-
- public String getScriptPath() {
- return scriptPath;
- }
+ // reset iPythonInterpreter to null as it is not available
+ iPythonInterpreter = null;
+ LOGGER.info("IPython is not available, use the native PythonInterpreter");
+ // Add matplotlib display hook
+ InterpreterGroup intpGroup = getInterpreterGroup();
+ if (intpGroup != null && intpGroup.getInterpreterHookRegistry() != null) {
+ try {
+ // just for unit test I believe (zjffdu)
+ registerHook(HookType.POST_EXEC_DEV.getName(), "__zeppelin__._displayhook()");
+ } catch (InvalidHookException e) {
+ throw new InterpreterException(e);
+ }
+ }
- private void copyFile(File out, String sourceFile) throws InterpreterException {
- ClassLoader classLoader = getClass().getClassLoader();
try {
- FileOutputStream outStream = new FileOutputStream(out);
- IOUtils.copy(
- classLoader.getResourceAsStream(sourceFile),
- outStream);
- outStream.close();
+ createGatewayServerAndStartScript();
} catch (IOException e) {
- throw new InterpreterException(e);
+ LOGGER.error("Fail to open PythonInterpreter", e);
+ throw new InterpreterException("Fail to open PythonInterpreter", e);
}
}
- private void createGatewayServerAndStartScript()
- throws UnknownHostException, InterpreterException {
- createPythonScript();
- if (System.getenv("ZEPPELIN_HOME") != null) {
- py4jLibPath = System.getenv("ZEPPELIN_HOME") + File.separator + ZEPPELIN_PY4JPATH;
- pythonLibPath = System.getenv("ZEPPELIN_HOME") + File.separator + ZEPPELIN_PYTHON_LIBS;
- } else {
- Path workingPath = Paths.get("..").toAbsolutePath();
- py4jLibPath = workingPath + File.separator + ZEPPELIN_PY4JPATH;
- pythonLibPath = workingPath + File.separator + ZEPPELIN_PYTHON_LIBS;
- }
-
- port = findRandomOpenPortOnAllLocalInterfaces();
+ // start gateway sever and start python process
+ private void createGatewayServerAndStartScript() throws IOException {
+ // start gateway server in JVM side
+ int port = RemoteInterpreterUtils.findRandomAvailablePortOnAllLocalInterfaces();
+ // use the FQDN as the server address instead of 127.0.0.1 so that python process in docker
+ // container can also connect to this gateway server.
+ String serverAddress = getLocalIP();
gatewayServer = new GatewayServer(this,
port,
GatewayServer.DEFAULT_PYTHON_PORT,
- InetAddress.getByName("0.0.0.0"),
- InetAddress.getByName("0.0.0.0"),
+ InetAddress.getByName(serverAddress),
+ InetAddress.getByName(serverAddress),
GatewayServer.DEFAULT_CONNECT_TIMEOUT,
GatewayServer.DEFAULT_READ_TIMEOUT,
- (List) null);
-
+ (List) null);;
gatewayServer.start();
+ LOGGER.info("Starting GatewayServer at " + serverAddress + ":" + port);
- // Run python shell
- String pythonCmd = getPythonCommand();
- CommandLine cmd = CommandLine.parse(pythonCmd);
-
- if (!pythonCmd.endsWith(".py")) {
- // PythonDockerInterpreter set pythoncmd with script
- cmd.addArgument(getScriptPath(), false);
+ // launch python process to connect to the gateway server in JVM side
+ createPythonScript();
+ String pythonExec = getPythonExec();
+ CommandLine cmd = CommandLine.parse(pythonExec);
+ if (!pythonExec.endsWith(".py")) {
+ // PythonDockerInterpreter set pythonExec with script
+ cmd.addArgument(pythonWorkDir + "/zeppelin_python.py", false);
}
+ cmd.addArgument(serverAddress, false);
cmd.addArgument(Integer.toString(port), false);
- cmd.addArgument(getLocalIp(), false);
executor = new DefaultExecutor();
- outputStream = new InterpreterOutputStream(LOG);
- PipedOutputStream ps = new PipedOutputStream();
- in = null;
- try {
- in = new PipedInputStream(ps);
- } catch (IOException e1) {
- throw new InterpreterException(e1);
- }
- ins = new BufferedWriter(new OutputStreamWriter(ps));
- input = new ByteArrayOutputStream();
-
- PumpStreamHandler streamHandler = new PumpStreamHandler(outputStream, outputStream, in);
+ outputStream = new InterpreterOutputStream(LOGGER);
+ PumpStreamHandler streamHandler = new PumpStreamHandler(outputStream);
executor.setStreamHandler(streamHandler);
executor.setWatchdog(new ExecuteWatchdog(ExecuteWatchdog.INFINITE_TIMEOUT));
- try {
- Map env = EnvironmentUtils.getProcEnvironment();
- if (!env.containsKey("PYTHONPATH")) {
- env.put("PYTHONPATH", py4jLibPath + File.pathSeparator + pythonLibPath);
- } else {
- env.put("PYTHONPATH", env.get("PYTHONPATH") + File.pathSeparator +
- py4jLibPath + File.pathSeparator + pythonLibPath);
- }
+ Map<String, String> env = setupPythonEnv();
+ LOGGER.info("Launching Python Process Command: " + cmd.getExecutable() +
+ " " + StringUtils.join(cmd.getArguments(), " "));
+ executor.execute(cmd, env, this);
+ pythonScriptRunning.set(true);
+ }
- logger.info("cmd = {}", cmd.toString());
- executor.execute(cmd, env, this);
- pythonscriptRunning = true;
- } catch (IOException e) {
- throw new InterpreterException(e);
+ private void createPythonScript() throws IOException {
+ // set java.io.tmpdir to /tmp on MacOS, because docker can not share the /var folder which will
+ // cause PythonDockerInterpreter fails.
+ // https://stackoverflow.com/questions/45122459/docker-mounts-denied-the-paths-are-not-shared-
+ // from-os-x-and-are-not-known
+ if (System.getProperty("os.name", "").contains("Mac")) {
+ System.setProperty("java.io.tmpdir", "/tmp");
+ }
+ this.pythonWorkDir = Files.createTempDir();
+ this.pythonWorkDir.deleteOnExit();
+ LOGGER.info("Create Python working dir: " + pythonWorkDir.getAbsolutePath());
+ copyResourceToPythonWorkDir("python/zeppelin_python.py", "zeppelin_python.py");
+ copyResourceToPythonWorkDir("python/zeppelin_context.py", "zeppelin_context.py");
+ copyResourceToPythonWorkDir("python/backend_zinline.py", "backend_zinline.py");
+ copyResourceToPythonWorkDir("python/mpl_config.py", "mpl_config.py");
+ copyResourceToPythonWorkDir("python/py4j-src-0.9.2.zip", "py4j-src-0.9.2.zip");
+ }
+
+ protected boolean useIPython() {
+ return this.iPythonInterpreter != null;
+ }
+
+ private String getLocalIP() {
+ // zeppelin.python.gatewayserver_address is only for unit test on travis.
+ // Because the FQDN would fail unit test on travis ci.
+ String gatewayserver_address =
+ properties.getProperty("zeppelin.python.gatewayserver_address");
+ if (gatewayserver_address != null) {
+ return gatewayserver_address;
}
try {
- input.write("import sys, getopt\n".getBytes());
- ins.flush();
- } catch (IOException e) {
- throw new InterpreterException(e);
+ return Inet4Address.getLocalHost().getHostAddress();
+ } catch (UnknownHostException e) {
+ LOGGER.warn("can't get local IP", e);
}
+ // fall back to loopback addreess
+ return "127.0.0.1";
}
- @Override
- public void open() throws InterpreterException {
- // try IPythonInterpreter first. If it is not available, we will fallback to the original
- // python interpreter implementation.
- iPythonInterpreter = getIPythonInterpreter();
- this.zeppelinContext = new PythonZeppelinContext(
- getInterpreterGroup().getInterpreterHookRegistry(),
- Integer.parseInt(getProperty("zeppelin.python.maxResult", "1000")));
- if (getProperty("zeppelin.python.useIPython", "true").equals("true") &&
- StringUtils.isEmpty(iPythonInterpreter.checkIPythonPrerequisite(getPythonBindPath()))) {
- try {
- iPythonInterpreter.open();
- LOG.info("IPython is available, Use IPythonInterpreter to replace PythonInterpreter");
- return;
- } catch (Exception e) {
- iPythonInterpreter = null;
- LOG.warn("Fail to open IPythonInterpreter", e);
+ private void copyResourceToPythonWorkDir(String srcResourceName,
+ String dstFileName) throws IOException {
+ FileOutputStream out = null;
+ try {
+ out = new FileOutputStream(pythonWorkDir.getAbsoluteFile() + "/" + dstFileName);
+ IOUtils.copy(
+ getClass().getClassLoader().getResourceAsStream(srcResourceName),
+ out);
+ } finally {
+ if (out != null) {
+ out.close();
}
}
+ }
- // reset iPythonInterpreter to null as it is not available
- iPythonInterpreter = null;
- LOG.info("IPython is not available, use the native PythonInterpreter");
- // Add matplotlib display hook
- InterpreterGroup intpGroup = getInterpreterGroup();
- if (intpGroup != null && intpGroup.getInterpreterHookRegistry() != null) {
- try {
- registerHook(HookType.POST_EXEC_DEV.getName(), "__zeppelin__._displayhook()");
- } catch (InvalidHookException e) {
- throw new InterpreterException(e);
- }
- }
- // Add matplotlib display hook
- try {
- createGatewayServerAndStartScript();
- } catch (UnknownHostException e) {
- throw new InterpreterException(e);
+ protected Map<String, String> setupPythonEnv() throws IOException {
+ Map<String, String> env = EnvironmentUtils.getProcEnvironment();
+ appendToPythonPath(env, pythonWorkDir.getAbsolutePath());
+ if (useBuiltinPy4j) {
+ appendToPythonPath(env, pythonWorkDir.getAbsolutePath() + "/py4j-src-0.9.2.zip");
}
+ LOGGER.info("PYTHONPATH: " + env.get("PYTHONPATH"));
+ return env;
}
- private IPythonInterpreter getIPythonInterpreter() {
- LazyOpenInterpreter lazy = null;
- IPythonInterpreter ipython = null;
- Interpreter p = getInterpreterInTheSameSessionByClassName(IPythonInterpreter.class.getName());
+ private void appendToPythonPath(Map<String, String> env, String path) {
+ if (!env.containsKey("PYTHONPATH")) {
+ env.put("PYTHONPATH", path);
+ } else {
+ env.put("PYTHONPATH", env.get("PYTHONPATH") + ":" + path);
+ }
+ }
- while (p instanceof WrappedInterpreter) {
- if (p instanceof LazyOpenInterpreter) {
- lazy = (LazyOpenInterpreter) p;
- }
- p = ((WrappedInterpreter) p).getInnerInterpreter();
+ // Run python script
+ // Choose python in the order of
+ // condaPythonExec > zeppelin.python
+ protected String getPythonExec() {
+ if (condaPythonExec != null) {
+ return condaPythonExec;
+ } else {
+ return getProperty("zeppelin.python", "python");
}
- ipython = (IPythonInterpreter) p;
- return ipython;
+ }
+
+ public File getPythonWorkDir() {
+ return pythonWorkDir;
}
@Override
@@ -286,54 +263,58 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
iPythonInterpreter.close();
return;
}
- pythonscriptRunning = false;
- pythonScriptInitialized = false;
-
- try {
- ins.flush();
- ins.close();
- input.flush();
- input.close();
- } catch (IOException e) {
- e.printStackTrace();
- }
+ pythonScriptRunning.set(false);
+ pythonScriptInitialized.set(false);
executor.getWatchdog().destroyProcess();
- new File(scriptPath).delete();
gatewayServer.shutdown();
- // wait until getStatements stop
- synchronized (statementSetNotifier) {
- try {
- statementSetNotifier.wait(1500);
- } catch (InterruptedException e) {
- }
- statementSetNotifier.notify();
- }
+ // reset these 2 monitors otherwise when you restart PythonInterpreter it would fails to execute
+ // python code as these 2 objects are in incorrect state.
+ statementSetNotifier = new Integer(0);
+ statementFinishedNotifier = new Integer(0);
+ }
+
+ private PythonInterpretRequest pythonInterpretRequest = null;
+ private Integer statementSetNotifier = new Integer(0);
+ private Integer statementFinishedNotifier = new Integer(0);
+ private String statementOutput = null;
+ private boolean statementError = false;
+
+ public void setPythonExec(String pythonExec) {
+ LOGGER.info("Set Python Command : {}", pythonExec);
+ this.condaPythonExec = pythonExec;
}
- PythonInterpretRequest pythonInterpretRequest = null;
/**
- * Result class of python interpreter
+ * Request send to Python Daemon
*/
public class PythonInterpretRequest {
public String statements;
+ public boolean isForCompletion;
- public PythonInterpretRequest(String statements) {
+ public PythonInterpretRequest(String statements, boolean isForCompletion) {
this.statements = statements;
+ this.isForCompletion = isForCompletion;
}
public String statements() {
return statements;
}
+
+ public boolean isForCompletion() {
+ return isForCompletion;
+ }
}
+ // called by Python Process
public PythonInterpretRequest getStatements() {
synchronized (statementSetNotifier) {
- while (pythonInterpretRequest == null && pythonscriptRunning && pythonScriptInitialized) {
+ while (pythonInterpretRequest == null) {
try {
statementSetNotifier.wait(1000);
} catch (InterruptedException e) {
+ e.printStackTrace();
}
}
PythonInterpretRequest req = pythonInterpretRequest;
@@ -342,65 +323,78 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
}
}
- String statementOutput = null;
- boolean statementError = false;
- Integer statementFinishedNotifier = new Integer(0);
-
+ // called by Python Process
public void setStatementsFinished(String out, boolean error) {
synchronized (statementFinishedNotifier) {
+ LOGGER.debug("Setting python statement output: " + out + ", error: " + error);
statementOutput = out;
statementError = error;
statementFinishedNotifier.notify();
}
}
- boolean pythonScriptInitialized = false;
- Integer pythonScriptInitializeNotifier = new Integer(0);
-
+ // called by Python Process
public void onPythonScriptInitialized(long pid) {
pythonPid = pid;
- synchronized (pythonScriptInitializeNotifier) {
- pythonScriptInitialized = true;
- pythonScriptInitializeNotifier.notifyAll();
+ synchronized (pythonScriptInitialized) {
+ LOGGER.debug("onPythonScriptInitialized is called");
+ pythonScriptInitialized.set(true);
+ pythonScriptInitialized.notifyAll();
}
}
+ // called by Python Process
public void appendOutput(String message) throws IOException {
+ LOGGER.debug("Output from python process: " + message);
outputStream.getInterpreterOutput().write(message);
}
- @Override
- public InterpreterResult interpret(String cmd, InterpreterContext contextInterpreter)
- throws InterpreterException {
- if (iPythonInterpreter != null) {
- return iPythonInterpreter.interpret(cmd, contextInterpreter);
- }
+ // used by subclass such as PySparkInterpreter to set JobGroup before executing spark code
+ protected void preCallPython(InterpreterContext context) {
- if (cmd == null || cmd.isEmpty()) {
- return new InterpreterResult(Code.SUCCESS, "");
+ }
+
+ // blocking call. Send python code to python process and get response
+ protected void callPython(PythonInterpretRequest request) {
+ synchronized (statementSetNotifier) {
+ this.pythonInterpretRequest = request;
+ statementOutput = null;
+ statementSetNotifier.notify();
}
- this.context = contextInterpreter;
+ synchronized (statementFinishedNotifier) {
+ while (statementOutput == null) {
+ try {
+ statementFinishedNotifier.wait(1000);
+ } catch (InterruptedException e) {
+ }
+ }
+ }
+ }
- zeppelinContext.setGui(context.getGui());
- zeppelinContext.setNoteGui(context.getNoteGui());
- zeppelinContext.setInterpreterContext(context);
+ @Override
+ public InterpreterResult interpret(String st, InterpreterContext context)
+ throws InterpreterException {
+ if (iPythonInterpreter != null) {
+ return iPythonInterpreter.interpret(st, context);
+ }
- if (!pythonscriptRunning) {
- return new InterpreterResult(Code.ERROR, "python process not running"
- + outputStream.toString());
+ if (!pythonScriptRunning.get()) {
+ return new InterpreterResult(Code.ERROR, "python process not running "
+ + outputStream.toString());
}
outputStream.setInterpreterOutput(context.out);
- synchronized (pythonScriptInitializeNotifier) {
+ synchronized (pythonScriptInitialized) {
long startTime = System.currentTimeMillis();
- while (pythonScriptInitialized == false
- && pythonscriptRunning
- && System.currentTimeMillis() - startTime < MAX_TIMEOUT_SEC * 1000) {
+ while (!pythonScriptInitialized.get()
+ && System.currentTimeMillis() - startTime < MAX_TIMEOUT_SEC * 1000) {
try {
- pythonScriptInitializeNotifier.wait(1000);
+ LOGGER.info("Wait for PythonScript initialized");
+ pythonScriptInitialized.wait(100);
} catch (InterruptedException e) {
+ e.printStackTrace();
}
}
}
@@ -413,59 +407,40 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
throw new InterpreterException(e);
}
- if (pythonscriptRunning == false) {
- // python script failed to initialize and terminated
- errorMessage.add(new InterpreterResultMessage(
- InterpreterResult.Type.TEXT, "failed to start python"));
- return new InterpreterResult(Code.ERROR, errorMessage);
- }
- if (pythonScriptInitialized == false) {
+ if (!pythonScriptInitialized.get()) {
// timeout. didn't get initialized message
errorMessage.add(new InterpreterResultMessage(
- InterpreterResult.Type.TEXT, "python is not responding"));
+ InterpreterResult.Type.TEXT, "Failed to initialize Python"));
return new InterpreterResult(Code.ERROR, errorMessage);
}
- pythonInterpretRequest = new PythonInterpretRequest(cmd);
- statementOutput = null;
-
- synchronized (statementSetNotifier) {
- statementSetNotifier.notify();
- }
+ BaseZeppelinContext z = getZeppelinContext();
+ z.setInterpreterContext(context);
+ z.setGui(context.getGui());
+ z.setNoteGui(context.getNoteGui());
+ InterpreterContext.set(context);
- synchronized (statementFinishedNotifier) {
- while (statementOutput == null) {
- try {
- statementFinishedNotifier.wait(1000);
- } catch (InterruptedException e) {
- }
- }
- }
+ preCallPython(context);
+ callPython(new PythonInterpretRequest(st, false));
if (statementError) {
return new InterpreterResult(Code.ERROR, statementOutput);
} else {
-
try {
context.out.flush();
} catch (IOException e) {
throw new InterpreterException(e);
}
-
return new InterpreterResult(Code.SUCCESS);
}
}
- public InterpreterContext getCurrentInterpreterContext() {
- return context;
- }
-
public void interrupt() throws IOException, InterpreterException {
if (pythonPid > -1) {
- logger.info("Sending SIGINT signal to PID : " + pythonPid);
+ LOGGER.info("Sending SIGINT signal to PID : " + pythonPid);
Runtime.getRuntime().exec("kill -SIGINT " + pythonPid);
} else {
- logger.warn("Non UNIX/Linux system, close the interpreter");
+ LOGGER.warn("Non UNIX/Linux system, close the interpreter");
close();
}
}
@@ -474,11 +449,12 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
public void cancel(InterpreterContext context) throws InterpreterException {
if (iPythonInterpreter != null) {
iPythonInterpreter.cancel(context);
+ return;
}
try {
interrupt();
} catch (IOException e) {
- e.printStackTrace();
+ LOGGER.error("Error", e);
}
}
@@ -495,114 +471,162 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
return 0;
}
- @Override
- public Scheduler getScheduler() {
- if (iPythonInterpreter != null) {
- return iPythonInterpreter.getScheduler();
- }
- return SchedulerFactory.singleton().createOrGetFIFOScheduler(
- PythonInterpreter.class.getName() + this.hashCode());
- }
@Override
public List<InterpreterCompletion> completion(String buf, int cursor,
- InterpreterContext interpreterContext) {
+ InterpreterContext interpreterContext)
+ throws InterpreterException {
if (iPythonInterpreter != null) {
return iPythonInterpreter.completion(buf, cursor, interpreterContext);
}
- return null;
- }
+ if (buf.length() < cursor) {
+ cursor = buf.length();
+ }
+ String completionString = getCompletionTargetString(buf, cursor);
+ String completionCommand = "__zeppelin_completion__.getCompletion('" + completionString + "')";
+ LOGGER.debug("completionCommand: " + completionCommand);
- public void setPythonCommand(String cmd) {
- logger.info("Set Python Command : {}", cmd);
- pythonCommand = cmd;
- }
+ pythonInterpretRequest = new PythonInterpretRequest(completionCommand, true);
+ statementOutput = null;
- private String getPythonCommand() {
- if (pythonCommand == null) {
- return getPythonBindPath();
- } else {
- return pythonCommand;
+ synchronized (statementSetNotifier) {
+ statementSetNotifier.notify();
}
- }
- public String getPythonBindPath() {
- String path = getProperty("zeppelin.python");
- if (path == null) {
- return DEFAULT_ZEPPELIN_PYTHON;
- } else {
- return path;
+ String[] completionList = null;
+ synchronized (statementFinishedNotifier) {
+ long startTime = System.currentTimeMillis();
+ while (statementOutput == null
+ && pythonScriptRunning.get()) {
+ try {
+ if (System.currentTimeMillis() - startTime > MAX_TIMEOUT_SEC * 1000) {
+ LOGGER.error("Python completion didn't have response for {}sec.", MAX_TIMEOUT_SEC);
+ break;
+ }
+ statementFinishedNotifier.wait(1000);
+ } catch (InterruptedException e) {
+ // not working
+ LOGGER.info("wait drop");
+ return new LinkedList<>();
+ }
+ }
+ if (statementError) {
+ return new LinkedList<>();
+ }
+ Gson gson = new Gson();
+ completionList = gson.fromJson(statementOutput, String[].class);
+ }
+ //end code for completion
+ if (completionList == null) {
+ return new LinkedList<>();
}
- }
- private Job getRunningJob(String paragraphId) {
- Job foundJob = null;
- Collection<Job> jobsRunning = getScheduler().getJobsRunning();
- for (Job job : jobsRunning) {
- if (job.getId().equals(paragraphId)) {
- foundJob = job;
- break;
- }
+ List<InterpreterCompletion> results = new LinkedList<>();
+ for (String name: completionList) {
+ results.add(new InterpreterCompletion(name, name, StringUtils.EMPTY));
}
- return foundJob;
+ return results;
}
- void bootStrapInterpreter(String file) throws IOException {
- BufferedReader bootstrapReader = new BufferedReader(
- new InputStreamReader(
- PythonInterpreter.class.getResourceAsStream(file)));
- String line = null;
- String bootstrapCode = "";
+ private String getCompletionTargetString(String text, int cursor) {
+ String[] completionSeqCharaters = {" ", "\n", "\t"};
+ int completionEndPosition = cursor;
+ int completionStartPosition = cursor;
+ int indexOfReverseSeqPostion = cursor;
- while ((line = bootstrapReader.readLine()) != null) {
- bootstrapCode += line + "\n";
+ String resultCompletionText = "";
+ String completionScriptText = "";
+ try {
+ completionScriptText = text.substring(0, cursor);
}
+ catch (Exception e) {
+ LOGGER.error(e.toString());
+ return null;
+ }
+ completionEndPosition = completionScriptText.length();
+
+ String tempReverseCompletionText = new StringBuilder(completionScriptText).reverse().toString();
+
+ for (String seqCharacter : completionSeqCharaters) {
+ indexOfReverseSeqPostion = tempReverseCompletionText.indexOf(seqCharacter);
+
+ if (indexOfReverseSeqPostion < completionStartPosition && indexOfReverseSeqPostion > 0) {
+ completionStartPosition = indexOfReverseSeqPostion;
+ }
- try {
- interpret(bootstrapCode, context);
- } catch (InterpreterException e) {
- throw new IOException(e);
}
- }
- public PythonZeppelinContext getZeppelinContext() {
- return zeppelinContext;
+ if (completionStartPosition == completionEndPosition) {
+ completionStartPosition = 0;
+ }
+ else
+ {
+ completionStartPosition = completionEndPosition - completionStartPosition;
+ }
+ resultCompletionText = completionScriptText.substring(
+ completionStartPosition , completionEndPosition);
+
+ return resultCompletionText;
}
- String getLocalIp() {
- try {
- return Inet4Address.getLocalHost().getHostAddress();
- } catch (UnknownHostException e) {
- logger.error("can't get local IP", e);
+ protected IPythonInterpreter getIPythonInterpreter() {
+ LazyOpenInterpreter lazy = null;
+ IPythonInterpreter iPython = null;
+ Interpreter p = getInterpreterInTheSameSessionByClassName(IPythonInterpreter.class.getName());
+
+ while (p instanceof WrappedInterpreter) {
+ if (p instanceof LazyOpenInterpreter) {
+ lazy = (LazyOpenInterpreter) p;
+ }
+ p = ((WrappedInterpreter) p).getInnerInterpreter();
}
- // fall back to loopback addreess
- return "127.0.0.1";
+ iPython = (IPythonInterpreter) p;
+ return iPython;
}
- private int findRandomOpenPortOnAllLocalInterfaces() {
- Integer port = -1;
- try (ServerSocket socket = new ServerSocket(0);) {
- port = socket.getLocalPort();
- socket.close();
- } catch (IOException e) {
- LOG.error("Can't find an open port", e);
+ protected BaseZeppelinContext createZeppelinContext() {
+ return new PythonZeppelinContext(
+ getInterpreterGroup().getInterpreterHookRegistry(),
+ Integer.parseInt(getProperty("zeppelin.python.maxResult", "1000")));
+ }
+
+ public BaseZeppelinContext getZeppelinContext() {
+ if (zeppelinContext == null) {
+ zeppelinContext = createZeppelinContext();
}
- return port;
+ return zeppelinContext;
}
- public int getMaxResult() {
- return maxResult;
+ protected void bootstrapInterpreter(String resourceName) throws IOException {
+ LOGGER.info("Bootstrap interpreter via " + resourceName);
+ String bootstrapCode =
+ IOUtils.toString(getClass().getClassLoader().getResourceAsStream(resourceName));
+ try {
+ InterpreterResult result = interpret(bootstrapCode, InterpreterContext.get());
+ if (result.code() != Code.SUCCESS) {
+ throw new IOException("Fail to run bootstrap script: " + resourceName);
+ }
+ } catch (InterpreterException e) {
+ throw new IOException(e);
+ }
}
@Override
public void onProcessComplete(int exitValue) {
- pythonscriptRunning = false;
- logger.info("python process terminated. exit code " + exitValue);
+ LOGGER.info("python process terminated. exit code " + exitValue);
+ pythonScriptRunning.set(false);
+ pythonScriptInitialized.set(false);
}
@Override
public void onProcessFailed(ExecuteException e) {
- pythonscriptRunning = false;
- logger.error("python process failed", e);
+ LOGGER.error("python process failed", e);
+ pythonScriptRunning.set(false);
+ pythonScriptInitialized.set(false);
+ }
+
+ // Called by Python Process, used for debugging purpose
+ public void logPythonOutput(String message) {
+ LOGGER.debug("Python Process Output: " + message);
}
}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/0a97446a/python/src/main/java/org/apache/zeppelin/python/PythonInterpreterPandasSql.java
----------------------------------------------------------------------
diff --git a/python/src/main/java/org/apache/zeppelin/python/PythonInterpreterPandasSql.java b/python/src/main/java/org/apache/zeppelin/python/PythonInterpreterPandasSql.java
index 54984c3..db65960 100644
--- a/python/src/main/java/org/apache/zeppelin/python/PythonInterpreterPandasSql.java
+++ b/python/src/main/java/org/apache/zeppelin/python/PythonInterpreterPandasSql.java
@@ -70,7 +70,7 @@ public class PythonInterpreterPandasSql extends Interpreter {
LOG.info("Bootstrap {} interpreter with {}", this.toString(), SQL_BOOTSTRAP_FILE_PY);
PythonInterpreter python = getPythonInterpreter();
- python.bootStrapInterpreter(SQL_BOOTSTRAP_FILE_PY);
+ python.bootstrapInterpreter(SQL_BOOTSTRAP_FILE_PY);
} catch (IOException e) {
LOG.error("Can't execute " + SQL_BOOTSTRAP_FILE_PY + " to import SQL dependencies", e);
}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/0a97446a/python/src/main/resources/python/backend_zinline.py
----------------------------------------------------------------------
diff --git a/python/src/main/resources/python/backend_zinline.py b/python/src/main/resources/python/backend_zinline.py
new file mode 100644
index 0000000..1c84699
--- /dev/null
+++ b/python/src/main/resources/python/backend_zinline.py
@@ -0,0 +1,317 @@
+# 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.
+
+# This file provides a static (non-interactive) matplotlib plotting backend
+# for zeppelin notebooks for use with the python/pyspark interpreters
+
+from __future__ import print_function
+
+import sys
+import uuid
+import warnings
+import base64
+from io import BytesIO
+try:
+ from StringIO import StringIO
+except ImportError:
+ from io import StringIO
+
+import mpl_config
+import matplotlib
+from matplotlib._pylab_helpers import Gcf
+from matplotlib.backends.backend_agg import new_figure_manager, FigureCanvasAgg
+from matplotlib.backend_bases import ShowBase, FigureManagerBase
+from matplotlib.figure import Figure
+
+########################################################################
+#
+# The following functions and classes are for pylab and implement
+# window/figure managers, etc...
+#
+########################################################################
+
+class Show(ShowBase):
+ """
+ A callable object that displays the figures to the screen. Valid kwargs
+ include figure width and height (in units supported by the div tag), block
+ (allows users to override blocking behavior regardless of whether or not
+ interactive mode is enabled, currently unused) and close (Implicitly call
+ matplotlib.pyplot.close('all') with each call to show()).
+ """
+ def __call__(self, close=None, block=None, **kwargs):
+ if close is None:
+ close = mpl_config.get('close')
+ try:
+ managers = Gcf.get_all_fig_managers()
+ if not managers:
+ return
+
+ # Tell zeppelin that the output will be html using the %html magic
+ # We want to do this only once to avoid seeing "%html" printed
+ # directly to the outout when multiple figures are displayed from
+ # one paragraph.
+ if mpl_config.get('angular'):
+ print('%angular')
+ else:
+ print('%html')
+
+ # Show all open figures
+ for manager in managers:
+ manager.show(**kwargs)
+ finally:
+ # This closes all the figures if close is set to True.
+ if close and Gcf.get_all_fig_managers():
+ Gcf.destroy_all()
+
+
+class FigureCanvasZInline(FigureCanvasAgg):
+ """
+ The canvas the figure renders into. Calls the draw and print fig
+ methods, creates the renderers, etc...
+ """
+ def get_bytes(self, **kwargs):
+ """
+ Get the byte representation of the figure.
+ Should only be used with jpg/png formats.
+ """
+ # Make sure format is correct
+ fmt = kwargs.get('format', mpl_config.get('format'))
+ if fmt == 'svg':
+ raise ValueError("get_bytes() does not support svg, use png or jpg")
+
+ # Express the image as bytes
+ buf = BytesIO()
+ self.print_figure(buf, **kwargs)
+ fmt = fmt.encode()
+ if sys.version_info >= (3, 4) and sys.version_info < (3, 5):
+ byte_str = bytes("data:image/%s;base64," %fmt, "utf-8")
+ else:
+ byte_str = b"data:image/%s;base64," %fmt
+ byte_str += base64.b64encode(buf.getvalue())
+
+ # Python3 forces all strings to default to unicode, but for raster image
+ # formats (eg png, jpg), we want to work with bytes. Thus this step is
+ # needed to ensure compatability for all python versions.
+ byte_str = byte_str.decode('ascii')
+ buf.close()
+ return byte_str
+
+ def get_svg(self, **kwargs):
+ """
+ Get the svg representation of the figure.
+ Should only be used with svg format.
+ """
+ # Make sure format is correct
+ fmt = kwargs.get('format', mpl_config.get('format'))
+ if fmt != 'svg':
+ raise ValueError("get_svg() does not support png or jpg, use svg")
+
+ # For SVG the data string has to be unicode, not bytes
+ buf = StringIO()
+ self.print_figure(buf, **kwargs)
+ svg_str = buf.getvalue()
+ buf.close()
+ return svg_str
+
+ def draw_idle(self, *args, **kwargs):
+ """
+ Called when the figure gets updated (eg through a plotting command).
+ This is overriden to allow open figures to be reshown after they
+ are updated when mpl_config.get('close') is False.
+ """
+ if not self._is_idle_drawing:
+ with self._idle_draw_cntx():
+ self.draw(*args, **kwargs)
+ draw_if_interactive()
+
+
+class FigureManagerZInline(FigureManagerBase):
+ """
+ Wrap everything up into a window for the pylab interface
+ """
+ def __init__(self, canvas, num):
+ FigureManagerBase.__init__(self, canvas, num)
+ self.fig_id = "figure_{0}".format(uuid.uuid4().hex)
+ self._shown = False
+
+ def angular_bind(self, **kwargs):
+ """
+ Bind figure data to Zeppelin's Angular Object Registry.
+ If mpl_config("angular") is True and PY4J is supported, this allows
+ for the possibility to interactively update a figure from a separate
+ paragraph without having to display it multiple times.
+ """
+ # This doesn't work for SVG so make sure it's not our format
+ fmt = kwargs.get('format', mpl_config.get('format'))
+ if fmt == 'svg':
+ return
+
+ # Get the figure data as a byte array
+ src = self.canvas.get_bytes(**kwargs)
+
+ # Flag to determine whether or not to use
+ # zeppelin's angular display system
+ angular = mpl_config.get('angular')
+
+ # ZeppelinContext instance (requires PY4J)
+ context = mpl_config.get('context')
+
+ # Finally we must ensure that automatic closing is set to False,
+ # as otherwise using the angular display system is pointless
+ close = mpl_config.get('close')
+
+ # If above conditions are met, bind the figure data to
+ # the Angular Object Registry.
+ if not close and angular:
+ if hasattr(context, 'angularBind'):
+ # Binding is performed through figure ID to ensure this works
+ # if multiple figures are open
+ context.angularBind(self.fig_id, src)
+
+ # Zeppelin will automatically replace this value even if it
+ # is updated from another pargraph thanks to the {{}} notation
+ src = "{{%s}}" %self.fig_id
+ else:
+ warnings.warn("Cannot bind figure to Angular Object Registry. "
+ "Check if PY4J is installed.")
+ return src
+
+ def angular_unbind(self):
+ """
+ Unbind figure from angular display system.
+ """
+ context = mpl_config.get('context')
+ if hasattr(context, 'angularUnbind'):
+ context.angularUnbind(self.fig_id)
+
+ def destroy(self):
+ """
+ Called when close=True or implicitly by pyplot.close().
+ Overriden to automatically clean up the angular object registry.
+ """
+ self.angular_unbind()
+
+ def show(self, **kwargs):
+ if not self._shown:
+ zdisplay(self.canvas.figure, **kwargs)
+ else:
+ self.canvas.draw_idle()
+ self.angular_bind(**kwargs)
+
+ self._shown = True
+
+
+def draw_if_interactive():
+ """
+ If interactive mode is on, this allows for updating properties of
+ the figure when each new plotting command is called.
+ """
+ manager = Gcf.get_active()
+ interactive = matplotlib.is_interactive()
+ angular = mpl_config.get('angular')
+
+ # Don't bother continuing if we aren't in interactive mode
+ # or if there are no active figures. Also pointless to continue
+ # in angular mode as we don't want to reshow the figure.
+ if not interactive or angular or manager is None:
+ return
+
+ # Allow for figure to be reshown if close is false since
+ # this function call implies that it has been updated
+ if not mpl_config.get('close'):
+ manager._shown = False
+
+
+def new_figure_manager(num, *args, **kwargs):
+ """
+ Create a new figure manager instance
+ """
+ # if a main-level app must be created, this (and
+ # new_figure_manager_given_figure) is the usual place to
+ # do it -- see backend_wx, backend_wxagg and backend_tkagg for
+ # examples. Not all GUIs require explicit instantiation of a
+ # main-level app (egg backend_gtk, backend_gtkagg) for pylab
+ FigureClass = kwargs.pop('FigureClass', Figure)
+ thisFig = FigureClass(*args, **kwargs)
+ return new_figure_manager_given_figure(num, thisFig)
+
+
+def new_figure_manager_given_figure(num, figure):
+ """
+ Create a new figure manager instance for the given figure.
+ """
+ canvas = FigureCanvasZInline(figure)
+ manager = FigureManagerZInline(canvas, num)
+ return manager
+
+
+########################################################################
+#
+# Backend specific functions
+#
+########################################################################
+
+def zdisplay(fig, **kwargs):
+ """
+ Publishes a matplotlib figure to the notebook paragraph output.
+ """
+ # kwargs can be width or height (in units supported by div tag)
+ width = kwargs.pop('width', 'auto')
+ height = kwargs.pop('height', 'auto')
+ fmt = kwargs.get('format', mpl_config.get('format'))
+
+ # Check if format is supported
+ supported_formats = mpl_config.get('supported_formats')
+ if fmt not in supported_formats:
+ raise ValueError("Unsupported format %s" %fmt)
+
+ # For SVG the data string has to be unicode, not bytes
+ if fmt == 'svg':
+ img = fig.canvas.get_svg(**kwargs)
+
+ # This is needed to ensure the SVG image is the correct size.
+ # We should find a better way to do this...
+ width = '{}px'.format(mpl_config.get('width'))
+ height = '{}px'.format(mpl_config.get('height'))
+ else:
+ # Express the image as bytes
+ src = fig.canvas.manager.angular_bind(**kwargs)
+ img = "<img src={src} style='width={width};height:{height}'>"
+ img = img.format(src=src, width=width, height=height)
+
+ # Print the image to the notebook paragraph via the %html magic
+ html = "<div style='width:{width};height:{height}'>{img}<div>"
+ print(html.format(width=width, height=height, img=img))
+
+def displayhook():
+ """
+ Called post paragraph execution if interactive mode is on
+ """
+ if matplotlib.is_interactive():
+ show()
+
+########################################################################
+#
+# Now just provide the standard names that backend.__init__ is expecting
+#
+########################################################################
+
+# Create a reference to the show function we are using. This is what actually
+# gets called by matplotlib.pyplot.show().
+show = Show()
+
+# Default FigureCanvas and FigureManager classes to use from the backend
+FigureCanvas = FigureCanvasZInline
+FigureManager = FigureManagerZInline
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/0a97446a/python/src/main/resources/python/mpl_config.py
----------------------------------------------------------------------
diff --git a/python/src/main/resources/python/mpl_config.py b/python/src/main/resources/python/mpl_config.py
new file mode 100644
index 0000000..5c60893
--- /dev/null
+++ b/python/src/main/resources/python/mpl_config.py
@@ -0,0 +1,105 @@
+# 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.
+
+# This module provides utitlites for users to configure the inline plotting
+# backend through a PyZeppelinContext instance (eg, through z.configure_mpl())
+
+import matplotlib
+
+def configure(**kwargs):
+ """
+ Generic configure function.
+ Usage: configure(prop1='foo', prop2='bar', ...)
+ Currently supported zeppelin-specific properties are:
+ interactive - If true show all figures without explicit call to show()
+ via a post-execute hook.
+ angular - If true, bind figures to angular display system.
+ close - If true, close all figures once shown.
+ width, height - Default width / height of the figure in pixels.
+ fontsize - Font size.
+ dpi - dpi of the figure.
+ fmt - Figure format
+ supported_formats - Supported Figure formats ()
+ context - ZeppelinContext instance (requires PY4J)
+ """
+ _config.update(**kwargs)
+
+ # Broadcast relevant changes to matplotlib RC
+ _on_config_change()
+
+
+def get(key):
+ """
+ Get the configuration info given a key
+ """
+ return _config[key]
+
+
+def _on_config_change():
+ # dpi
+ dpi = _config['dpi']
+
+ # For older versions of matplotlib, savefig.dpi is not synced with
+ # figure.dpi by default
+ matplotlib.rcParams['figure.dpi'] = dpi
+ if matplotlib.__version__ < '2.0.0':
+ matplotlib.rcParams['savefig.dpi'] = dpi
+
+ # Width and height
+ width = float(_config['width']) / dpi
+ height = float(_config['height']) / dpi
+ matplotlib.rcParams['figure.figsize'] = (width, height)
+
+ # Font size
+ fontsize = _config['fontsize']
+ matplotlib.rcParams['font.size'] = fontsize
+
+ # Default Figure Format
+ fmt = _config['format']
+ supported_formats = _config['supported_formats']
+ if fmt not in supported_formats:
+ raise ValueError("Unsupported format %s" %fmt)
+
+ if matplotlib.__version__ < '1.2.0':
+ matplotlib.rcParams.update({'savefig.format': fmt})
+ else:
+ matplotlib.rcParams['savefig.format'] = fmt
+
+ # Interactive mode
+ interactive = _config['interactive']
+ matplotlib.interactive(interactive)
+
+
+def _init_config():
+ dpi = matplotlib.rcParams['figure.dpi']
+ if matplotlib.__version__ < '1.2.0':
+ matplotlib.rcParams.update({'savefig.format': 'png'})
+ fmt = matplotlib.rcParams['savefig.format']
+ width, height = matplotlib.rcParams['figure.figsize']
+ fontsize = matplotlib.rcParams['font.size']
+ _config['dpi'] = dpi
+ _config['format'] = fmt
+ _config['width'] = width*dpi
+ _config['height'] = height*dpi
+ _config['fontsize'] = fontsize
+ _config['close'] = True
+ _config['interactive'] = matplotlib.is_interactive()
+ _config['angular'] = False
+ _config['supported_formats'] = ['png', 'jpg', 'svg']
+ _config['context'] = None
+
+
+_config = {}
+_init_config()
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/0a97446a/python/src/main/resources/python/py4j-src-0.9.2.zip
----------------------------------------------------------------------
diff --git a/python/src/main/resources/python/py4j-src-0.9.2.zip b/python/src/main/resources/python/py4j-src-0.9.2.zip
new file mode 100644
index 0000000..8ceb15c
Binary files /dev/null and b/python/src/main/resources/python/py4j-src-0.9.2.zip differ