You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "gaogaotiantian (via GitHub)" <gi...@apache.org> on 2026/02/11 01:10:15 UTC
[PR] [SPARK-55476] Refactor broadcast variable protocol [spark]
gaogaotiantian opened a new pull request, #54258:
URL: https://github.com/apache/spark/pull/54258
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Refactored the broadcast variable protocol from random value being sent to a json-based structured message.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
It's part of the effort to make protocol between JVM and the workers more stable and easier to understand.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Tried it on some relevant tests locally and they passed. Need CI to validate the rest.
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
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-55476][PYTHON] Refactor broadcast variable protocol [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #54258:
URL: https://github.com/apache/spark/pull/54258#discussion_r2801553149
##########
python/pyspark/worker_util.py:
##########
@@ -155,39 +155,42 @@ def setup_spark_files(infile: IO) -> None:
def setup_broadcasts(infile: IO) -> None:
"""
Set up broadcasted variables.
+ {
+ "conn_info": int | str | None,
+ "auth_secret": str | None,
+ "broadcast_variables": [
+ {
+ "bid": int,
+ "path": str | None,
+ }
+ ]
+ }
"""
if not is_remote_only():
from pyspark.core.broadcast import Broadcast, _broadcastRegistry
- # fetch names and values of broadcast variables
- needs_broadcast_decryption_server = read_bool(infile)
- num_broadcast_variables = read_int(infile)
- if needs_broadcast_decryption_server:
+ data = json.loads(utf8_deserializer.loads(infile))
Review Comment:
My point is that the benefit is not super compelling. If we plan to refactor the whole protocol or sth, yeah probably we should go ahead. But doing this alone doesn't look worthwhile to me.
If we do want to refactor the whole protocol, we should better have a bigger picture.
--
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-55476][PYTHON] Refactor broadcast variable protocol [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #54258:
URL: https://github.com/apache/spark/pull/54258#discussion_r2801374018
##########
python/pyspark/worker_util.py:
##########
@@ -155,39 +155,42 @@ def setup_spark_files(infile: IO) -> None:
def setup_broadcasts(infile: IO) -> None:
"""
Set up broadcasted variables.
+ {
+ "conn_info": int | str | None,
+ "auth_secret": str | None,
+ "broadcast_variables": [
+ {
+ "bid": int,
+ "path": str | None,
+ }
+ ]
+ }
"""
if not is_remote_only():
from pyspark.core.broadcast import Broadcast, _broadcastRegistry
- # fetch names and values of broadcast variables
- needs_broadcast_decryption_server = read_bool(infile)
- num_broadcast_variables = read_int(infile)
- if needs_broadcast_decryption_server:
+ data = json.loads(utf8_deserializer.loads(infile))
Review Comment:
If that's the case, I am not super supportive of this change. This could impact jobs like Structured Streaming (with micro batches) or ML jobs that disable `spark.python.worker.reuse` (which happen often in practice to work around any problem by having long living daemon worker). Considering the overhead vs benefit, I would prefer to just leave it.
--
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-55476][PYTHON] Refactor broadcast variable protocol [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #54258:
URL: https://github.com/apache/spark/pull/54258#discussion_r2800737420
##########
python/pyspark/worker_util.py:
##########
@@ -155,39 +155,42 @@ def setup_spark_files(infile: IO) -> None:
def setup_broadcasts(infile: IO) -> None:
"""
Set up broadcasted variables.
+ {
+ "conn_info": int | str | None,
+ "auth_secret": str | None,
+ "broadcast_variables": [
+ {
+ "bid": int,
+ "path": str | None,
+ }
+ ]
+ }
"""
if not is_remote_only():
from pyspark.core.broadcast import Broadcast, _broadcastRegistry
- # fetch names and values of broadcast variables
- needs_broadcast_decryption_server = read_bool(infile)
- num_broadcast_variables = read_int(infile)
- if needs_broadcast_decryption_server:
+ data = json.loads(utf8_deserializer.loads(infile))
Review Comment:
I did not quite get the question. We are not using it for debugging. This is part of the protocol. We need this from JVM.
--
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-55476][PYTHON] Refactor broadcast variable protocol [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #54258:
URL: https://github.com/apache/spark/pull/54258#discussion_r2801534033
##########
python/pyspark/worker_util.py:
##########
@@ -155,39 +155,42 @@ def setup_spark_files(infile: IO) -> None:
def setup_broadcasts(infile: IO) -> None:
"""
Set up broadcasted variables.
+ {
+ "conn_info": int | str | None,
+ "auth_secret": str | None,
+ "broadcast_variables": [
+ {
+ "bid": int,
+ "path": str | None,
+ }
+ ]
+ }
"""
if not is_remote_only():
from pyspark.core.broadcast import Broadcast, _broadcastRegistry
- # fetch names and values of broadcast variables
- needs_broadcast_decryption_server = read_bool(infile)
- num_broadcast_variables = read_int(infile)
- if needs_broadcast_decryption_server:
+ data = json.loads(utf8_deserializer.loads(infile))
Review Comment:
Performance impact is a big red herring. This change introduced two kinds of "overhead":
* CPU time to encode/decode json
* Extra bytes through the JVM/worker network (it's on the same machine)
Decoding a small json string takes about 1us. It's probably on the same range on scala side. Local network runs at least 10Gbps, an extra 100 bytes takes about 0.1us.
That's the overhead we introduce for every UDF run.
Currently, without reuse-worker, each worker takes about a few hundred ms to spawn. I made an optimization a few weeks ago that eliminated 100-200ms per spawn for reused worker and no one even notice it.
1us is 0.01% of 100ms. That's literally nothing. If we care about 1us, we have serious issues with our current UDF. I can get a lot of 1us from our current code if that's what we need to make our protocol more stable.
--
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-55476][PYTHON] Refactor broadcast variable protocol [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #54258:
URL: https://github.com/apache/spark/pull/54258#issuecomment-4526988266
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
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-55476][PYTHON] Refactor broadcast variable protocol [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #54258: [SPARK-55476][PYTHON] Refactor broadcast variable protocol
URL: https://github.com/apache/spark/pull/54258
--
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-55476][PYTHON] Refactor broadcast variable protocol [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #54258:
URL: https://github.com/apache/spark/pull/54258#discussion_r2801244383
##########
python/pyspark/worker_util.py:
##########
@@ -155,39 +155,42 @@ def setup_spark_files(infile: IO) -> None:
def setup_broadcasts(infile: IO) -> None:
"""
Set up broadcasted variables.
+ {
+ "conn_info": int | str | None,
+ "auth_secret": str | None,
+ "broadcast_variables": [
+ {
+ "bid": int,
+ "path": str | None,
+ }
+ ]
+ }
"""
if not is_remote_only():
from pyspark.core.broadcast import Broadcast, _broadcastRegistry
- # fetch names and values of broadcast variables
- needs_broadcast_decryption_server = read_bool(infile)
- num_broadcast_variables = read_int(infile)
- if needs_broadcast_decryption_server:
+ data = json.loads(utf8_deserializer.loads(infile))
Review Comment:
Sorry use what? This is part of the JVM <-> Python worker protocol. We are not adding any new features. JVM used to send broadcast variable information integer by integer (with some strings in the middle) to Python. Now instead of that raw fragile protocol, we send all of the broadcast variable information in a JSON string.
--
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-55476][PYTHON] Refactor broadcast variable protocol [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #54258:
URL: https://github.com/apache/spark/pull/54258#discussion_r2797584727
##########
python/pyspark/worker_util.py:
##########
@@ -155,39 +155,42 @@ def setup_spark_files(infile: IO) -> None:
def setup_broadcasts(infile: IO) -> None:
"""
Set up broadcasted variables.
+ {
+ "conn_info": int | str | None,
+ "auth_secret": str | None,
+ "broadcast_variables": [
+ {
+ "bid": int,
+ "path": str | None,
+ }
+ ]
+ }
"""
if not is_remote_only():
from pyspark.core.broadcast import Broadcast, _broadcastRegistry
- # fetch names and values of broadcast variables
- needs_broadcast_decryption_server = read_bool(infile)
- num_broadcast_variables = read_int(infile)
- if needs_broadcast_decryption_server:
+ data = json.loads(utf8_deserializer.loads(infile))
Review Comment:
How would we use this info for debugging? I feel like you can just log it here instead of JSON ser/de
--
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-55476][PYTHON] Refactor broadcast variable protocol [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #54258:
URL: https://github.com/apache/spark/pull/54258#discussion_r2801180857
##########
python/pyspark/worker_util.py:
##########
@@ -155,39 +155,42 @@ def setup_spark_files(infile: IO) -> None:
def setup_broadcasts(infile: IO) -> None:
"""
Set up broadcasted variables.
+ {
+ "conn_info": int | str | None,
+ "auth_secret": str | None,
+ "broadcast_variables": [
+ {
+ "bid": int,
+ "path": str | None,
+ }
+ ]
+ }
"""
if not is_remote_only():
from pyspark.core.broadcast import Broadcast, _broadcastRegistry
- # fetch names and values of broadcast variables
- needs_broadcast_decryption_server = read_bool(infile)
- num_broadcast_variables = read_int(infile)
- if needs_broadcast_decryption_server:
+ data = json.loads(utf8_deserializer.loads(infile))
Review Comment:
Would you mind describing how we're going to use this in the PR description?
--
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-55476][PYTHON] Refactor broadcast variable protocol [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #54258:
URL: https://github.com/apache/spark/pull/54258#discussion_r2801365830
##########
python/pyspark/worker_util.py:
##########
@@ -155,39 +155,42 @@ def setup_spark_files(infile: IO) -> None:
def setup_broadcasts(infile: IO) -> None:
"""
Set up broadcasted variables.
+ {
+ "conn_info": int | str | None,
+ "auth_secret": str | None,
+ "broadcast_variables": [
+ {
+ "bid": int,
+ "path": str | None,
+ }
+ ]
+ }
"""
if not is_remote_only():
from pyspark.core.broadcast import Broadcast, _broadcastRegistry
- # fetch names and values of broadcast variables
- needs_broadcast_decryption_server = read_bool(infile)
- num_broadcast_variables = read_int(infile)
- if needs_broadcast_decryption_server:
+ data = json.loads(utf8_deserializer.loads(infile))
Review Comment:
I am trying to understand why we need this. Is this to purely make the protocol more stable?
--
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-55476][PYTHON] Refactor broadcast variable protocol [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #54258:
URL: https://github.com/apache/spark/pull/54258#discussion_r2801583031
##########
python/pyspark/worker_util.py:
##########
@@ -155,39 +155,42 @@ def setup_spark_files(infile: IO) -> None:
def setup_broadcasts(infile: IO) -> None:
"""
Set up broadcasted variables.
+ {
+ "conn_info": int | str | None,
+ "auth_secret": str | None,
+ "broadcast_variables": [
+ {
+ "bid": int,
+ "path": str | None,
+ }
+ ]
+ }
"""
if not is_remote_only():
from pyspark.core.broadcast import Broadcast, _broadcastRegistry
- # fetch names and values of broadcast variables
- needs_broadcast_decryption_server = read_bool(infile)
- num_broadcast_variables = read_int(infile)
- if needs_broadcast_decryption_server:
+ data = json.loads(utf8_deserializer.loads(infile))
Review Comment:
This is part of the effort to refactor the whole protocol. I'm doing it piece by piece so eventually we can have a structured message from JVM. Ideally a multi-phased message. All the initialization (probably the UDF definition) should be sent in a single message. The message should be relatively resistant to new changes. For example it won't stuck if we decide to add something new. Or it should report a clear error when the message is not following the protocol.
json is good in a sense that, if we decide to add something else to broadcast variable protocol, it's easy. If we did it wrong, we can find it quickly too. Otherwise we had to be really careful about where to insert the `read_long` and if we did not do it correctly, the worker could stuck at an arbitrary point.
This refactor actually eliminated some unnecessary fields. Our old protocol is too fragile that no one is willing to touch it. I want to gradually convert it to a more structured way. A single switch is a bit too dangerous.
--
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