This is an automated email from the ASF dual-hosted git repository.
ParkGyeongTae pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/zeppelin.git
The following commit(s) were added to refs/heads/master by this push:
new e1d96d65d6 [ZEPPELIN-6457] Extract shared Python ZeppelinContext
helpers for PySpark and PyFlink
e1d96d65d6 is described below
commit e1d96d65d6c9314039a73d485d11e28d52f7b133
Author: DONGHOON LEE <[email protected]>
AuthorDate: Sun Aug 9 21:05:19 2026 +0900
[ZEPPELIN-6457] Extract shared Python ZeppelinContext helpers for PySpark
and PyFlink
### What is this PR for?
PySpark and PyFlink currently define their `ZeppelinContext` subclasses
separately in the classic Python and IPython bootstrap resources, even though
the implementations are effectively the same for each backend.
This PR moves the PySpark and PyFlink implementations into the shared
`zeppelin_context.py` resource and reuses them from both bootstrap paths.
Backend-specific behavior remains unchanged:
* PySpark handles Spark `DataFrame` objects through `_jdf`.
* PyFlink handles Flink `Table` objects through `_j_table`.
* PyFlink retains support for the `stream_type` argument.
* Other object types fall back to `PyZeppelinContext.show()`.
* `IPySparkZeppelinContext` and `IPyFlinkZeppelinContext` remain available
as compatibility aliases.
No new dependencies are introduced.
### What type of PR is it?
Refactoring
### Todos
* [x] Move the PySpark ZeppelinContext implementation to the shared resource
* [x] Move the PyFlink ZeppelinContext implementation to the shared resource
* [x] Reuse the shared implementations from Python and IPython bootstraps
* [x] Keep the existing IPython class names as compatibility aliases
* [x] Verify Spark and Flink interpreter packaging
* [x] Verify PySpark `z.show(DataFrame)` behavior
### What is the Jira issue?
https://issues.apache.org/jira/browse/ZEPPELIN-6457
### How should this be tested?
The following checks passed successfully:
```bash
./mvnw test \
-pl python \
-Dtest=PythonInterpreterTest#testBackendZeppelinContextsAvailable \
-DfailIfNoTests=false \
-Dmaven.gitcommitid.skip=true
```
Result: 1 test passed.
```bash
PYSPARK_PYTHON="$(command -v python3.12)" \
PYSPARK_DRIVER_PYTHON="$(command -v python3.12)" \
./mvnw test \
-pl spark/interpreter \
--am \
-Pspark-3.5 \
-Pspark-scala-2.12 \
-Dtest=PySparkInterpreterTest#testPySpark \
-Dsurefire.failIfNoSpecifiedTests=false \
-Dmaven.gitcommitid.skip=true
```
Results:
* `PySparkInterpreterTest#testPySpark` passed.
* Existing ScalaTest suite: 18 tests passed.
* `z.show(DataFrame)` was exercised successfully.
```bash
./mvnw package \
-pl spark/interpreter \
--am \
-Pspark-3.5 \
-Pspark-scala-2.12 \
-DskipTests \
-Dmaven.gitcommitid.skip=true
```
Result: 12-module reactor build succeeded.
```bash
./mvnw package \
-pl flink/flink-scala-2.12 \
--am \
-Pflink-1.20 \
-DskipTests \
-Dmaven.gitcommitid.skip=true
```
Result: 13-module reactor build succeeded.
The generated Spark and Flink interpreter JARs were also verified to
contain the shared context resource and the corresponding classic and IPython
bootstrap resources.
`git diff --check` also passes.
### Screenshots (if appropriate)
Not applicable. This PR does not change the user interface.
### Questions
* Do the license files need to be updated? No.
* Are there breaking changes for older versions? No. The existing IPython
context names remain available as compatibility aliases.
* Does this need documentation? No. There are no user-facing behavior or
configuration changes.
Closes #5405 from move-hoon/ZEPPELIN-6457-python-context.
Signed-off-by: ParkGyeongTae <[email protected]>
---
.../src/main/resources/python/zeppelin_ipyflink.py | 17 +-------------
.../src/main/resources/python/zeppelin_pyflink.py | 18 +--------------
.../src/main/resources/python/zeppelin_context.py | 27 ++++++++++++++++++++++
.../zeppelin/python/PythonInterpreterTest.java | 15 ++++++++++++
.../src/main/resources/python/zeppelin_ipyspark.py | 14 +----------
.../src/main/resources/python/zeppelin_pyspark.py | 15 +-----------
6 files changed, 46 insertions(+), 60 deletions(-)
diff --git
a/flink/flink-scala-2.12/src/main/resources/python/zeppelin_ipyflink.py
b/flink/flink-scala-2.12/src/main/resources/python/zeppelin_ipyflink.py
index 18bab2111f..6db9907077 100644
--- a/flink/flink-scala-2.12/src/main/resources/python/zeppelin_ipyflink.py
+++ b/flink/flink-scala-2.12/src/main/resources/python/zeppelin_ipyflink.py
@@ -54,19 +54,4 @@ if not intp.isAfterFlink114():
else:
st_env = StreamTableEnvironment(intp.getJavaStreamTableEnvironment())
-class IPyFlinkZeppelinContext(PyZeppelinContext):
-
- def __init__(self, z, gateway):
- super(IPyFlinkZeppelinContext, self).__init__(z, gateway)
-
- def show(self, obj, **kwargs):
- from pyflink.table import Table
- if isinstance(obj, Table):
- if 'stream_type' in kwargs:
- self.z.show(obj._j_table, kwargs['stream_type'], kwargs)
- else:
- print(self.z.showData(obj._j_table))
- else:
- super(IPyFlinkZeppelinContext, self).show(obj, **kwargs)
-
-z = __zeppelin__ = IPyFlinkZeppelinContext(intp.getZeppelinContext(), gateway)
+z = __zeppelin__ = PyFlinkZeppelinContext(intp.getZeppelinContext(), gateway)
diff --git
a/flink/flink-scala-2.12/src/main/resources/python/zeppelin_pyflink.py
b/flink/flink-scala-2.12/src/main/resources/python/zeppelin_pyflink.py
index 88d1de6c59..9d76c2633d 100644
--- a/flink/flink-scala-2.12/src/main/resources/python/zeppelin_pyflink.py
+++ b/flink/flink-scala-2.12/src/main/resources/python/zeppelin_pyflink.py
@@ -44,23 +44,7 @@ else:
st_env = StreamTableEnvironment(intp.getJavaStreamTableEnvironment())
-from zeppelin_context import PyZeppelinContext
-
-#TODO(zjffdu) merge it with IPyFlinkZeppelinContext
-class PyFlinkZeppelinContext(PyZeppelinContext):
-
- def __init__(self, z, gateway):
- super(PyFlinkZeppelinContext, self).__init__(z, gateway)
-
- def show(self, obj, **kwargs):
- from pyflink.table import Table
- if isinstance(obj, Table):
- if 'stream_type' in kwargs:
- self.z.show(obj._j_table, kwargs['stream_type'], kwargs)
- else:
- print(self.z.showData(obj._j_table))
- else:
- super(PyFlinkZeppelinContext, self).show(obj, **kwargs)
+from zeppelin_context import PyFlinkZeppelinContext
z = __zeppelin__ = PyFlinkZeppelinContext(intp.getZeppelinContext(), gateway)
__zeppelin__._setup_matplotlib()
diff --git a/python/src/main/resources/python/zeppelin_context.py
b/python/src/main/resources/python/zeppelin_context.py
index 8223966d40..4325d9a4cc 100644
--- a/python/src/main/resources/python/zeppelin_context.py
+++ b/python/src/main/resources/python/zeppelin_context.py
@@ -287,3 +287,30 @@ class PyZeppelinContext(object):
matplotlib.use('Agg')
warnings.warn("Unable to load inline matplotlib backend, "
"falling back to Agg")
+
+
+class PySparkZeppelinContext(PyZeppelinContext):
+
+ def show(self, obj, **kwargs):
+ from pyspark.sql import DataFrame
+ if isinstance(obj, DataFrame):
+ print(self.z.showData(obj._jdf))
+ else:
+ super(PySparkZeppelinContext, self).show(obj, **kwargs)
+
+
+class PyFlinkZeppelinContext(PyZeppelinContext):
+
+ def show(self, obj, **kwargs):
+ from pyflink.table import Table
+ if isinstance(obj, Table):
+ if 'stream_type' in kwargs:
+ self.z.show(obj._j_table, kwargs['stream_type'], kwargs)
+ else:
+ print(self.z.showData(obj._j_table))
+ else:
+ super(PyFlinkZeppelinContext, self).show(obj, **kwargs)
+
+
+IPySparkZeppelinContext = PySparkZeppelinContext
+IPyFlinkZeppelinContext = PyFlinkZeppelinContext
diff --git
a/python/src/test/java/org/apache/zeppelin/python/PythonInterpreterTest.java
b/python/src/test/java/org/apache/zeppelin/python/PythonInterpreterTest.java
index 7d0ad4f0e2..839ab6aad3 100644
--- a/python/src/test/java/org/apache/zeppelin/python/PythonInterpreterTest.java
+++ b/python/src/test/java/org/apache/zeppelin/python/PythonInterpreterTest.java
@@ -82,6 +82,21 @@ public class PythonInterpreterTest extends
BasePythonInterpreterTest {
intpGroup.close();
}
+ @Test
+ void testBackendZeppelinContextsAvailable() throws InterpreterException,
IOException {
+ InterpreterContext context = getInterpreterContext();
+ InterpreterResult result = interpreter.interpret(
+ "from zeppelin_context import PySparkZeppelinContext,
PyFlinkZeppelinContext, " +
+ "IPySparkZeppelinContext, IPyFlinkZeppelinContext\n" +
+ "print('%s %s' % (IPySparkZeppelinContext is
PySparkZeppelinContext, " +
+ "IPyFlinkZeppelinContext is PyFlinkZeppelinContext))",
+ context);
+
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+ assertEquals("True True",
+ context.out.toInterpreterResultMessage().get(0).getData().trim());
+ }
+
@Override
public void testCodeCompletion() throws InterpreterException, IOException,
InterruptedException {
super.testCodeCompletion();
diff --git a/spark/interpreter/src/main/resources/python/zeppelin_ipyspark.py
b/spark/interpreter/src/main/resources/python/zeppelin_ipyspark.py
index 958802ccdb..94da41b9d3 100644
--- a/spark/interpreter/src/main/resources/python/zeppelin_ipyspark.py
+++ b/spark/interpreter/src/main/resources/python/zeppelin_ipyspark.py
@@ -64,19 +64,7 @@ if intp.isAfterSpark33():
else:
sqlContext = sqlc = __zSqlc__ = __zSpark__._wrapped
-class IPySparkZeppelinContext(PyZeppelinContext):
-
- def __init__(self, z, gateway):
- super(IPySparkZeppelinContext, self).__init__(z, gateway)
-
- def show(self, obj, **kwargs):
- from pyspark.sql import DataFrame
- if isinstance(obj, DataFrame):
- print(self.z.showData(obj._jdf))
- else:
- super(IPySparkZeppelinContext, self).show(obj, **kwargs)
-
-z = __zeppelin__ = IPySparkZeppelinContext(intp.getZeppelinContext(), gateway)
+z = __zeppelin__ = PySparkZeppelinContext(intp.getZeppelinContext(), gateway)
# add jars to path
import sys
diff --git a/spark/interpreter/src/main/resources/python/zeppelin_pyspark.py
b/spark/interpreter/src/main/resources/python/zeppelin_pyspark.py
index 52788ccdb3..fb3b6ee856 100644
--- a/spark/interpreter/src/main/resources/python/zeppelin_pyspark.py
+++ b/spark/interpreter/src/main/resources/python/zeppelin_pyspark.py
@@ -56,20 +56,7 @@ if intp.isAfterSpark33():
else:
sqlContext = sqlc = __zSqlc__ = __zSpark__._wrapped
-from zeppelin_context import PyZeppelinContext
-
-#TODO(zjffdu) merge it with IPySparkZeppelinContext
-class PySparkZeppelinContext(PyZeppelinContext):
-
- def __init__(self, z, gateway):
- super(PySparkZeppelinContext, self).__init__(z, gateway)
-
- def show(self, obj, **kwargs):
- from pyspark.sql import DataFrame
- if isinstance(obj, DataFrame):
- print(self.z.showData(obj._jdf))
- else:
- super(PySparkZeppelinContext, self).show(obj, **kwargs)
+from zeppelin_context import PySparkZeppelinContext
z = __zeppelin__ = PySparkZeppelinContext(intp.getZeppelinContext(), gateway)
__zeppelin__._setup_matplotlib()
