You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "HyukjinKwon (via GitHub)" <gi...@apache.org> on 2025/04/15 04:09:05 UTC
[PR] Spark 51688 followup [spark]
HyukjinKwon opened a new pull request, #50587:
URL: https://github.com/apache/spark/pull/50587
### What changes were proposed in this pull request?
This PR is a followup of https://github.com/apache/spark/pull/50466 that enables Unix Domain Sockets in Accumulators as well. This will be the last PR to complete UDS in PySpark.
### Why are the changes needed?
This was not handled in the original PR as it was a bit complicated. Was separated out. See the original PR.
### Does this PR introduce _any_ user-facing change?
See https://github.com/apache/spark/pull/50466
### How was this patch tested?
CI in this PR with enabling it by default. Also will set up a scheduled build through https://github.com/apache/spark/pull/50585
### Was this patch authored or co-authored using generative AI tooling?
No.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51688][PYTHON][FOLLOW-UP] Implement UDS in Accumulators [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #50587:
URL: https://github.com/apache/spark/pull/50587#issuecomment-2804151229
Merged to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51688][PYTHON][FOLLOW-UP] Implement UDS in Accumulators [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50587:
URL: https://github.com/apache/spark/pull/50587#discussion_r2050505986
##########
core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala:
##########
@@ -717,35 +718,50 @@ class BytesToString extends org.apache.spark.api.java.function.Function[Array[By
* collects a list of pickled strings that we pass to Python through a socket.
*/
private[spark] class PythonAccumulatorV2(
- @transient private val serverHost: String,
- private val serverPort: Int,
- private val secretToken: String)
+ @transient private val serverHost: Option[String],
+ private val serverPort: Option[Int],
+ private val secretToken: Option[String],
+ @transient private val socketPath: Option[String])
extends CollectionAccumulator[Array[Byte]] with Logging {
- Utils.checkHost(serverHost)
+ // Unix domain socket
+ def this(socketPath: String) = this(None, None, None, Some(socketPath))
+ // TPC socket
+ def this(serverHost: String, serverPort: Int, secretToken: String) = this(
+ Some(serverHost), Some(serverPort), Some(secretToken), None)
+
+ serverHost.foreach(Utils.checkHost)
val bufferSize = SparkEnv.get.conf.get(BUFFER_SIZE)
+ val isUnixDomainSock = SparkEnv.get.conf.get(PYTHON_UNIX_DOMAIN_SOCKET_ENABLED)
/**
* We try to reuse a single Socket to transfer accumulator updates, as they are all added
* by the DAGScheduler's single-threaded RpcEndpoint anyway.
*/
- @transient private var socket: Socket = _
+ @transient private var socket: SocketChannel = _
- private def openSocket(): Socket = synchronized {
- if (socket == null || socket.isClosed) {
- socket = new Socket(serverHost, serverPort)
- logInfo(log"Connected to AccumulatorServer at host: ${MDC(HOST, serverHost)}" +
- log" port: ${MDC(PORT, serverPort)}")
+ private def openSocket(): SocketChannel = synchronized {
+ if (socket == null || !socket.isOpen) {
+ if (isUnixDomainSock) {
+ socket = SocketChannel.open(UnixDomainSocketAddress.of(socketPath.get))
+ logInfo(log"Connected to AccumulatorServer at socket: ${MDC(SOCKET_ADDRESS, serverHost)}")
Review Comment:
serverHost.get
##########
core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala:
##########
@@ -717,35 +718,50 @@ class BytesToString extends org.apache.spark.api.java.function.Function[Array[By
* collects a list of pickled strings that we pass to Python through a socket.
*/
private[spark] class PythonAccumulatorV2(
- @transient private val serverHost: String,
- private val serverPort: Int,
- private val secretToken: String)
+ @transient private val serverHost: Option[String],
+ private val serverPort: Option[Int],
+ private val secretToken: Option[String],
+ @transient private val socketPath: Option[String])
extends CollectionAccumulator[Array[Byte]] with Logging {
- Utils.checkHost(serverHost)
+ // Unix domain socket
+ def this(socketPath: String) = this(None, None, None, Some(socketPath))
+ // TPC socket
+ def this(serverHost: String, serverPort: Int, secretToken: String) = this(
+ Some(serverHost), Some(serverPort), Some(secretToken), None)
+
+ serverHost.foreach(Utils.checkHost)
val bufferSize = SparkEnv.get.conf.get(BUFFER_SIZE)
+ val isUnixDomainSock = SparkEnv.get.conf.get(PYTHON_UNIX_DOMAIN_SOCKET_ENABLED)
/**
* We try to reuse a single Socket to transfer accumulator updates, as they are all added
* by the DAGScheduler's single-threaded RpcEndpoint anyway.
*/
- @transient private var socket: Socket = _
+ @transient private var socket: SocketChannel = _
- private def openSocket(): Socket = synchronized {
- if (socket == null || socket.isClosed) {
- socket = new Socket(serverHost, serverPort)
- logInfo(log"Connected to AccumulatorServer at host: ${MDC(HOST, serverHost)}" +
- log" port: ${MDC(PORT, serverPort)}")
+ private def openSocket(): SocketChannel = synchronized {
+ if (socket == null || !socket.isOpen) {
+ if (isUnixDomainSock) {
+ socket = SocketChannel.open(UnixDomainSocketAddress.of(socketPath.get))
+ logInfo(log"Connected to AccumulatorServer at socket: ${MDC(SOCKET_ADDRESS, serverHost)}")
+ } else {
+ socket = SocketChannel.open(new InetSocketAddress(serverHost.get, serverPort.get))
+ logInfo(log"Connected to AccumulatorServer at host: ${MDC(HOST, serverHost)}" +
Review Comment:
serverHost.get
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51688][PYTHON][FOLLOW-UP] Implement UDS in Accumulators [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #50587: [SPARK-51688][PYTHON][FOLLOW-UP] Implement UDS in Accumulators
URL: https://github.com/apache/spark/pull/50587
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51688][PYTHON][FOLLOW-UP] Implement UDS in Accumulators [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #50587:
URL: https://github.com/apache/spark/pull/50587#issuecomment-2803941043
Tests with the configuration enabled: https://github.com/HyukjinKwon/spark/actions/runs/14461375675/job/40554555642
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51688][PYTHON][FOLLOW-UP] Implement UDS in Accumulators [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50587:
URL: https://github.com/apache/spark/pull/50587#discussion_r2050506132
##########
core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala:
##########
@@ -717,35 +718,50 @@ class BytesToString extends org.apache.spark.api.java.function.Function[Array[By
* collects a list of pickled strings that we pass to Python through a socket.
*/
private[spark] class PythonAccumulatorV2(
- @transient private val serverHost: String,
- private val serverPort: Int,
- private val secretToken: String)
+ @transient private val serverHost: Option[String],
+ private val serverPort: Option[Int],
+ private val secretToken: Option[String],
+ @transient private val socketPath: Option[String])
extends CollectionAccumulator[Array[Byte]] with Logging {
- Utils.checkHost(serverHost)
+ // Unix domain socket
+ def this(socketPath: String) = this(None, None, None, Some(socketPath))
+ // TPC socket
+ def this(serverHost: String, serverPort: Int, secretToken: String) = this(
+ Some(serverHost), Some(serverPort), Some(secretToken), None)
+
+ serverHost.foreach(Utils.checkHost)
val bufferSize = SparkEnv.get.conf.get(BUFFER_SIZE)
+ val isUnixDomainSock = SparkEnv.get.conf.get(PYTHON_UNIX_DOMAIN_SOCKET_ENABLED)
/**
* We try to reuse a single Socket to transfer accumulator updates, as they are all added
* by the DAGScheduler's single-threaded RpcEndpoint anyway.
*/
- @transient private var socket: Socket = _
+ @transient private var socket: SocketChannel = _
- private def openSocket(): Socket = synchronized {
- if (socket == null || socket.isClosed) {
- socket = new Socket(serverHost, serverPort)
- logInfo(log"Connected to AccumulatorServer at host: ${MDC(HOST, serverHost)}" +
- log" port: ${MDC(PORT, serverPort)}")
+ private def openSocket(): SocketChannel = synchronized {
+ if (socket == null || !socket.isOpen) {
+ if (isUnixDomainSock) {
+ socket = SocketChannel.open(UnixDomainSocketAddress.of(socketPath.get))
+ logInfo(log"Connected to AccumulatorServer at socket: ${MDC(SOCKET_ADDRESS, serverHost)}")
+ } else {
+ socket = SocketChannel.open(new InetSocketAddress(serverHost.get, serverPort.get))
+ logInfo(log"Connected to AccumulatorServer at host: ${MDC(HOST, serverHost)}" +
+ log" port: ${MDC(PORT, serverPort)}")
Review Comment:
serverPort.get
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51688][PYTHON][FOLLOW-UP] Implement UDS in Accumulators [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #50587:
URL: https://github.com/apache/spark/pull/50587#discussion_r2043544921
##########
core/src/main/scala/org/apache/spark/internal/config/Python.scala:
##########
@@ -77,7 +77,7 @@ private[spark] object Python {
"to Spark Classic and Spark Connect server.")
.version("4.1.0")
.booleanConf
- .createWithDefault(sys.env.get("PYSPARK_UDS_MODE").contains("true"))
+ .createWithDefault(true)
Review Comment:
I will disable this back when I merge. It will be disabled by default.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51688][PYTHON][FOLLOW-UP] Implement UDS in Accumulators [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #50587:
URL: https://github.com/apache/spark/pull/50587#issuecomment-2803731737
cc @ueshin
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51688][PYTHON][FOLLOW-UP] Implement UDS in Accumulators [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #50587:
URL: https://github.com/apache/spark/pull/50587#issuecomment-2817385311
Thanks, made a Pr at https://github.com/apache/spark/pull/50648
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org