[SPARK-55476][PYTHON] Refactor broadcast variable protocol - #54258
[SPARK-55476][PYTHON] Refactor broadcast variable protocol#54258gaogaotiantian wants to merge 1 commit into
Conversation
a094831 to
7f600b5
Compare
HyukjinKwon
changed the title
| 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)) |
There was a problem hiding this comment.
How would we use this info for debugging? I feel like you can just log it here instead of JSON ser/de
There was a problem hiding this 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.
There was a problem hiding this comment.
Would you mind describing how we're going to use this in the PR description?
There was a problem hiding this 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.
There was a problem hiding this comment.
I am trying to understand why we need this. Is this to purely make the protocol more stable?
There was a problem hiding this 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.
There was a problem hiding this 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.
There was a problem hiding this 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.
There was a problem hiding this 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.
|
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. |
2 participants
What changes were proposed in this pull request?
Refactored the broadcast variable protocol from random value being sent to a json-based structured message.
Why are the changes needed?
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?
No.
How was this patch tested?
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?
No.