Skip to content

Commit 8d20090

Browse files
dhama-shashank-meeshodhama-shashank-meesho
committed
[ZEPPELIN-XXXX] Add Spark Connect interpreter
Adds a new spark-connect module that enables Apache Zeppelin to connect to remote Spark clusters via the Spark Connect gRPC protocol (Spark 3.5+). This eliminates the need for a local Spark installation on the Zeppelin host. The module includes four interpreters: - %spark-connect: SQL execution (default) - %spark-connect.sql: SQL with optional concurrent scheduler - %spark-connect.pyspark: PySpark via Py4j bridge - %spark-connect.ipyspark: IPython variant of PySpark Key design decisions: - Per-user session quota enforcement to prevent resource exhaustion - Per-notebook fair ReentrantLock for sequential query execution - Dependency shading to avoid Netty/gRPC conflicts - Token-based authentication and SSL support - Python bridge via Py4j for shared Java SparkSession Includes comprehensive unit tests (SparkConnectUtilsTest) and integration tests (gated by SPARK_CONNECT_TEST_REMOTE environment variable) for SparkConnectInterpreter, SparkConnectSqlInterpreter, and PySparkConnectInterpreter.
1 parent f6d37d2 commit 8d20090

17 files changed

Lines changed: 3678 additions & 9 deletions

‎pom.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,7 @@
6262
<module>zeppelin-jupyter-interpreter-shaded</module>
6363
<module>groovy</module>
6464
<module>spark</module>
65+
<module>spark-connect</module>
6566
<module>spark-submit</module>
6667
<module>markdown</module>
6768
<module>mongodb</module>

‎scripts/docker/zeppelin-interpreter/env_python_3_with_R.yml‎

Lines changed: 35 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -4,29 +4,55 @@ channels:
44
- defaults
55
dependencies:
66
- python >=3.9,<3.10
7-
- pyspark=3.5.3
7+
- pyspark=3.5
88
- pycodestyle
9-
- scipy
9+
# --- Core data libraries ---
10+
- pandas
1011
- numpy
12+
- scipy
13+
- pyarrow
14+
# --- Spark Connect protocol ---
1115
- grpcio
1216
- protobuf
17+
# --- HTTP / networking ---
18+
- requests
19+
- urllib3
20+
# --- File format support ---
21+
- openpyxl
22+
- xlrd
23+
- pyyaml
24+
- tabulate
25+
# --- GCP access ---
26+
- google-cloud-storage
27+
- google-auth
28+
- gcsfs
29+
# --- Visualization ---
30+
- matplotlib
31+
- seaborn
32+
- plotly
33+
- plotnine
34+
- altair
35+
- vega_datasets
36+
- hvplot
37+
# --- SQL on pandas ---
1338
- pandasql
39+
# --- ML ---
40+
- scikit-learn
41+
- xgboost
42+
# --- IPython / kernel ---
1443
- ipython
1544
- ipykernel
1645
- jupyter_client
17-
- hvplot
18-
- plotnine
19-
- seaborn
46+
# --- Data connectors ---
2047
- intake
2148
- intake-parquet
2249
- intake-xarray
23-
- altair
24-
- vega_datasets
25-
- plotly
50+
# --- pip-only packages ---
2651
- pip
2752
- pip:
28-
# works for regular pip packages
2953
- bkzep==0.6.1
54+
- delta-spark==3.2.1
55+
# --- R support ---
3056
- r-base=3
3157
- r-data.table
3258
- r-evaluate

‎spark-connect/README.md‎

Lines changed: 253 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,253 @@
1+
# Spark Connect Interpreter for Apache Zeppelin
2+
3+
## What is Spark Connect?
4+
5+
Spark Connect (Spark 3.5+) is a new client-server architecture for Apache Spark that decouples the Spark client from the Spark cluster. Unlike the traditional `spark` interpreter which requires running Spark in the same JVM process, Spark Connect is a **thin gRPC client** that communicates with a remote Spark cluster via the `sc://host:port` connection string.
6+
7+
## Why Use Spark Connect?
8+
9+
- **No local Spark installation** — Zeppelin doesn't need the full Spark distribution on its host
10+
- **Remote cluster support** — Connect to any Spark 3.5+ cluster over the network
11+
- **Token authentication** — Support for token-based auth and SSL
12+
- **Multi-user isolation** — Per-user session quotas prevent resource exhaustion
13+
- **Cleaner deployments** — Simpler Docker images, reduced memory footprint
14+
15+
## Differences from the Legacy Spark Interpreter
16+
17+
| Feature | Spark Interpreter | Spark Connect Interpreter |
18+
|---------|-------------------|---------------------------|
19+
| **Architecture** | In-process SparkContext | Remote gRPC client |
20+
| **Installation** | Requires full Spark on host | Only Spark Connect client JAR needed |
21+
| **Scala support** | Yes (via embedded Scala interpreter) | No (client-only protocol) |
22+
| **R support** | Yes (SparkR) | No (not supported) |
23+
| **ZeppelinContext** | Full support (`z.show()`, Angular) | Returns `null` |
24+
| **Multi-user** | Global shared SparkContext | Isolated sessions per user |
25+
| **Session quota** | Not enforced | Per-user quota (default: 5) |
26+
27+
## Prerequisites
28+
29+
1. **Spark 3.5.x cluster** running the Spark Connect server
30+
```bash
31+
# Start a Spark Connect server on port 15002
32+
spark-shell --master <cluster-url> --conf spark.connect.grpc.binding.port=15002
33+
```
34+
35+
2. **Python 3.x** (for PySpark/IPySpark support)
36+
37+
## Configuration Properties
38+
39+
The Spark Connect interpreter supports the following configuration properties in the Zeppelin UI:
40+
41+
### Connection Settings
42+
43+
| Property | Type | Default | Description |
44+
|----------|------|---------|-------------|
45+
| `zeppelin.spark.connect.url` | string | `sc://localhost:15002` | Spark Connect server URL |
46+
| `zeppelin.spark.connect.token` | string | `` | Optional token for authentication (redacted in logs) |
47+
| `zeppelin.spark.connect.use_ssl` | checkbox | false | Enable SSL/TLS for connection |
48+
| `zeppelin.spark.connect.user_id` | string | `` | User ID to report to Spark Connect server |
49+
50+
### Session Management
51+
52+
| Property | Type | Default | Description |
53+
|----------|------|---------|-------------|
54+
| `zeppelin.spark.connect.maxSessionsPerUser` | number | 5 | Maximum concurrent sessions per user |
55+
56+
### Execution
57+
58+
| Property | Type | Default | Description |
59+
|----------|------|---------|-------------|
60+
| `zeppelin.spark.maxResult` | number | 10000 | Maximum result rows to fetch |
61+
| `zeppelin.spark.concurrentSQL` | checkbox | false | Allow concurrent SQL execution (within notebook) |
62+
| `zeppelin.spark.concurrentSQL.max` | number | 10 | Max concurrent SQL threads |
63+
64+
### PySpark
65+
66+
| Property | Type | Default | Description |
67+
|----------|------|---------|-------------|
68+
| `zeppelin.python` | string | `python` | Python executable path |
69+
| `zeppelin.pyspark.useIPython` | checkbox | true | Use IPython if available |
70+
71+
## Usage
72+
73+
### SQL Mode (Default)
74+
75+
```sql
76+
%spark-connect
77+
78+
SELECT * FROM my_table LIMIT 10
79+
```
80+
81+
### SQL with Concurrent Execution
82+
83+
```sql
84+
%spark-connect.sql
85+
86+
-- This uses the concurrentSQL scheduler
87+
SELECT COUNT(*) FROM large_table
88+
```
89+
90+
### PySpark
91+
92+
```python
93+
%spark-connect.pyspark
94+
95+
df = spark.sql("SELECT * FROM my_table")
96+
df.show()
97+
```
98+
99+
### IPython PySpark
100+
101+
```python
102+
%spark-connect.ipyspark
103+
104+
# Full IPython REPL with Spark
105+
df = spark.sql("SELECT COUNT(*) FROM table")
106+
df.collect()
107+
```
108+
109+
## Architecture
110+
111+
### SparkConnectInterpreter
112+
- Core interpreter managing the remote Spark session
113+
- Enforces per-user session quota via `ConcurrentHashMap<String, Integer>`
114+
- Uses `NotebookLockManager` for per-notebook sequential execution
115+
- Delegates SQL parsing to `SqlSplitter` for multi-statement support
116+
117+
### SparkConnectSqlInterpreter
118+
- SQL-only frontend with optional concurrent scheduler
119+
- Shares the Spark session with `SparkConnectInterpreter`
120+
121+
### PySparkConnectInterpreter
122+
- Bridges Java `SparkSession` to Python via Py4j
123+
- Uses custom Python wrapper classes (`SparkConnectDataFrame`, `SparkConnectSession`)
124+
- Supports same Python executable resolution as Spark's own PySpark
125+
126+
### IPySparkConnectInterpreter
127+
- IPython variant of PySpark
128+
- Uses the same shared session model
129+
130+
### SparkConnectUtils
131+
- Stateless utilities for:
132+
- Building Spark Connect URIs with token/SSL/user_id params
133+
- Formatting DataFrames as Zeppelin `%table` output
134+
- Streaming large result sets to avoid memory overflow
135+
136+
### NotebookLockManager
137+
- Per-notebook `ReentrantLock` registry (fair FIFO ordering)
138+
- Ensures sequential query execution within a single notebook
139+
- Prevents concurrent modifications to shared notebook state
140+
141+
## Session Isolation and Multi-User Support
142+
143+
Each user gets **isolated Spark sessions** tracked in a global `ConcurrentHashMap<String, Integer>`:
144+
- Username extracted from Zeppelin auth (falls back to `"anonymous"`)
145+
- Per-user quota enforced (`maxSessionsPerUser`, default 5)
146+
- Prevents runaway session proliferation
147+
148+
Within a notebook, a fair `ReentrantLock` ensures:
149+
- Only one query executes at a time (even with `concurrentSQL=true`)
150+
- FIFO ordering prevents starvation
151+
152+
## Testing
153+
154+
### Unit Tests (No Spark Server Required)
155+
156+
Tests for `SparkConnectUtils` utility class:
157+
```bash
158+
mvn test -pl spark-connect -Dtest=SparkConnectUtilsTest
159+
```
160+
161+
### Integration Tests (Requires Spark Connect Server)
162+
163+
Full interpreter tests with a live Spark server:
164+
```bash
165+
SPARK_CONNECT_TEST_REMOTE=sc://localhost:15002 \
166+
mvn test -pl spark-connect
167+
```
168+
169+
Only integration tests are executed when `SPARK_CONNECT_TEST_REMOTE` is set.
170+
171+
## Limitations
172+
173+
1. **No Scala interpreter** — Spark Connect is a client-only protocol; embedded Scala REPL not supported
174+
2. **No R support** — `%spark.r` and `%spark.ir` not available
175+
3. **No ZeppelinContext** — `z.show()`, Angular widgets, and other Zeppelin-specific features return `null`
176+
4. **Spark 3.5.x only** — The gRPC protocol is version-locked to Spark 3.5
177+
5. **No progress tracking** — Job progress API always returns 0
178+
179+
## Dependency Shading
180+
181+
The module uses Maven Shade Plugin to relocate conflicting dependencies:
182+
- `io.netty` → `org.apache.zeppelin.spark.connect.io.netty`
183+
- `com.google` → `org.apache.zeppelin.spark.connect.com.google`
184+
- `io.grpc` → `org.apache.zeppelin.spark.connect.io.grpc`
185+
186+
This prevents classpath conflicts with Zeppelin Server's own Netty and other interpreters.
187+
188+
## Examples
189+
190+
### Connect to a Remote Spark Cluster
191+
192+
Configure in Zeppelin UI:
193+
- **URL**: `sc://spark-server.example.com:15002`
194+
- **Token**: `your-auth-token` (if required)
195+
- **Use SSL**: Enable if cluster uses TLS
196+
197+
### Run Multi-Statement SQL
198+
199+
```sql
200+
%spark-connect
201+
202+
CREATE OR REPLACE TEMP VIEW my_view AS
203+
SELECT * FROM source_table WHERE year = 2024;
204+
205+
SELECT COUNT(*) FROM my_view;
206+
```
207+
208+
### PySpark with Pandas
209+
210+
```python
211+
%spark-connect.pyspark
212+
213+
# Create a Spark DataFrame and convert to Pandas
214+
import pandas as pd
215+
df_spark = spark.sql("SELECT * FROM my_table")
216+
df_pandas = df_spark.toPandas()
217+
print(df_pandas.head())
218+
```
219+
220+
### Inspect DataFrame Schema
221+
222+
```python
223+
%spark-connect.pyspark
224+
225+
df = spark.sql("SELECT * FROM events LIMIT 1")
226+
df.explain() # Logical and physical plan
227+
df.printSchema() # Column names and types
228+
```
229+
230+
## Troubleshooting
231+
232+
### Connection Refused
233+
234+
- Ensure Spark Connect server is running: `spark-shell --master <url> --conf spark.connect.grpc.binding.port=15002`
235+
- Verify network connectivity and firewall rules
236+
- Check `zeppelin.spark.connect.url` configuration
237+
238+
### Authentication Failures
239+
240+
- Verify `zeppelin.spark.connect.token` matches server token requirements
241+
- Enable `zeppelin.spark.connect.use_ssl` if cluster uses TLS
242+
243+
### Out of Memory
244+
245+
- Use `zeppelin.spark.maxResult` to limit rows fetched
246+
- Use `spark.sql(...).limit(n)` in queries to reduce data transfer
247+
- Enable `zeppelin.spark.connect.use_ssl` to stream results instead of collecting
248+
249+
## References
250+
251+
- [Apache Spark Connect Documentation](https://spark.apache.org/docs/latest/spark-connect-overview.html)
252+
- [Spark Connect Protocol](https://spark.apache.org/docs/latest/spark-connect-introduction.html)
253+
- [Zeppelin Interpreter Development](https://zeppelin.apache.org/docs/latest/usage/interpreter/interpreter_binding_mode.html)

0 commit comments

Comments
 (0)