You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2025/03/06 23:09:08 UTC
(spark) branch branch-4.0 updated: [SPARK-51416][CONNECT] Remove SPARK_CONNECT_MODE when starting Spark Connect server
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new b824dfed967e [SPARK-51416][CONNECT] Remove SPARK_CONNECT_MODE when starting Spark Connect server
b824dfed967e is described below
commit b824dfed967ebeee96035d6aa6fbbdf19e7beb6d
Author: Hyukjin Kwon <gu...@apache.org>
AuthorDate: Fri Mar 7 08:08:50 2025 +0900
[SPARK-51416][CONNECT] Remove SPARK_CONNECT_MODE when starting Spark Connect server
### What changes were proposed in this pull request?
This PR proposes to remove `SPARK_CONNECT_MODE` environment variable when starting Spark Connect server. So SparkSubmit thinks no remote is set in order to start the regular session.
### Why are the changes needed?
To make Spark Connect version of distribution works with `bin/pyspark`. Currently it fails as below:
```
py4j.protocol.Py4JJavaError: An error occurred while calling None.org.apache.spark.api.java.JavaSparkContext.
: java.lang.ClassNotFoundException: org.apache.spark.sql.connect.SparkConnectPlugin
at java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:445)
at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:592)
at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:525)
at java.base/java.lang.Class.forName0(Native Method)
at java.base/java.lang.Class.forName(Class.java:467)
at org.apache.spark.util.SparkClassUtils.classForName(SparkClassUtils.scala:41)
at org.apache.spark.util.SparkClassUtils.classForName$(SparkClassUtils.scala:36)
at org.apache.spark.util.Utils$.classForName(Utils.scala:99)
at org.apache.spark.util.Utils$.$anonfun$loadExtensions$1(Utils.scala:2828)
at scala.collection.StrictOptimizedIterableOps.flatMap(StrictOptimizedIterableOps.scala:118)
at scala.collection.StrictOptimizedIterableOps.flatMap$(StrictOptimizedIterableOps.scala:105)
at scala.collection.immutable.ArraySeq.flatMap(ArraySeq.scala:35)
at org.apache.spark.util.Utils$.loadExtensions(Utils.scala:2826)
at org.apache.spark.internal.plugin.PluginContainer$.apply(PluginContainer.scala:210)
at org.apache.spark.internal.plugin.PluginContainer$.apply(PluginContainer.scala:196)
at org.apache.spark.SparkContext.<init>(SparkContext.scala:588)
at org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:59)
at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:500)
at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:481)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:247)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374)
at py4j.Gateway.invoke(Gateway.java:238)
at py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:80)
at py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:69)
at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:184)
at py4j.ClientServerConnection.run(ClientServerConnection.java:108)
at java.base/java.lang.Thread.run(Thread.java:840)
```
`bin/spark-shell` also fails due to the same reason.
### Does this PR introduce _any_ user-facing change?
No, the main change has not been released yet.
### How was this patch tested?
Manually with:
```
SPARK_CONNECT_MODE=1 ./bin/pyspark
SPARK_CONNECT_MODE=1 ./bin/spark-shell
```
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #50180 from HyukjinKwon/SPARK-51416.
Authored-by: Hyukjin Kwon <gu...@apache.org>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
(cherry picked from commit e4cc116791dc3b8c8dfcd9421867e7ddb8f12e70)
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
python/pyspark/sql/connect/session.py | 9 +++++++--
.../main/scala/org/apache/spark/sql/connect/SparkSession.scala | 3 ++-
2 files changed, 9 insertions(+), 3 deletions(-)
diff --git a/python/pyspark/sql/connect/session.py b/python/pyspark/sql/connect/session.py
index 79677262eb16..76c6050e7615 100644
--- a/python/pyspark/sql/connect/session.py
+++ b/python/pyspark/sql/connect/session.py
@@ -1074,11 +1074,14 @@ class SparkSession:
overwrite_conf["spark.connect.grpc.binding.port"] = "0"
origin_remote = os.environ.get("SPARK_REMOTE", None)
+ origin_connect_mode = os.environ.get("SPARK_CONNECT_MODE", None)
try:
+ # So SparkSubmit thinks no remote is set in order to
+ # start the regular PySpark session.
if origin_remote is not None:
- # So SparkSubmit thinks no remote is set in order to
- # start the regular PySpark session.
del os.environ["SPARK_REMOTE"]
+ if origin_connect_mode is not None:
+ del os.environ["SPARK_CONNECT_MODE"]
# The regular PySpark session is registered as an active session
# so would not be garbage-collected.
@@ -1096,6 +1099,8 @@ class SparkSession:
finally:
if origin_remote is not None:
os.environ["SPARK_REMOTE"] = origin_remote
+ if origin_connect_mode is not None:
+ os.environ["SPARK_CONNECT_MODE"] = origin_connect_mode
else:
raise PySparkRuntimeError(
errorClass="SESSION_OR_CONTEXT_EXISTS",
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala
index 0af7c7b6d97a..b8e4a49261be 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala
@@ -744,7 +744,7 @@ object SparkSession extends SparkSessionCompanion with Logging {
lazy val isAPIModeConnect =
Option(System.getProperty(org.apache.spark.sql.SparkSessionBuilder.API_MODE_KEY))
.getOrElse("classic")
- .toLowerCase(Locale.ROOT) == "connect"
+ .toLowerCase(Locale.ROOT) == "connect" || System.getenv("SPARK_CONNECT_MODE") == "1"
val remoteString = sparkOptions
.get("spark.remote")
.orElse(Option(System.getProperty("spark.remote"))) // Set from Spark Submit
@@ -778,6 +778,7 @@ object SparkSession extends SparkSessionCompanion with Logging {
val pb = new ProcessBuilder(args: _*)
// So don't exclude spark-sql jar in classpath
pb.environment().remove(SparkConnectClient.SPARK_REMOTE)
+ pb.environment().remove("SPARK_CONNECT_MODE")
pb.environment().put("SPARK_IDENT_STRING", serverId)
pb.environment().put("HOSTNAME", "local")
pb.environment().put("SPARK_CONNECT_AUTHENTICATE_TOKEN", token)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org