[SPARK-59621][CORE] Serialize StatusUpdate manually to avoid Enumeration/BigDecimal overhead - #58894
Open
david-mollitor-db wants to merge 1 commit into
Open
david-mollitor-db wants to merge 1 commit into
david-mollitor-db wants to merge 1 commit into
Conversation
…ion/BigDecimal overhead `CoarseGrainedClusterMessages.StatusUpdate` is sent from executor to driver on every task state change (`RUNNING` at start, `FINISHED`/`FAILED`/`KILLED` at end) -- roughly two or more per task, on the driver's RPC intake path. An empty-payload `StatusUpdate` Java-serializes to 1713 bytes, and two fields account for ~75% of it: `state` (a Scala `Enumeration` value, 642 bytes) and `taskCpus` (a `BigDecimal`, 638 bytes). A fresh `ObjectOutputStream` per RPC means no class-descriptor caching, so every message pays this in full. A Scala `Enumeration.Value` serializes a reference to its enclosing `Enumeration` object, dragging in the whole `TaskState` object; `scala.math.BigDecimal` drags in `java.math.BigDecimal` + `BigInteger` + `MathContext` + `RoundingMode` descriptors. The actual payload is ~31 bytes. Make `StatusUpdate` `Externalizable` with a compact manual encoding, mirroring `UpdateBlockInfo` in the same message family and reusing `TaskDescription`'s exact wire form for the fractional-CPU `BigDecimal` (`CpuAmount.toDisplayString` / `CpuAmount.normalize`): `state` as one byte, `taskCpus` as its normalized decimal string, `data` via `SerializableBuffer`'s existing channel-based serialization (no extra copy for large results), and `resources` as a size-prefixed nested map. This shrinks the empty-payload message from 1713 bytes to 191 bytes (~9x smaller), cutting steady serialization/GC/bandwidth on a per-task-frequency control message. It is behavior-preserving (all fields round-trip; fractional CPUs exactly) and is not a throughput claim. Verified with a new CoarseGrainedClusterMessagesSuite (round-trips all fields, every TaskState, fractional taskCpus exactly, empty payload, and a size guard). Co-authored-by: Isaac <no-reply@databricks.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
CoarseGrainedClusterMessages.StatusUpdateis sent from executor to driver on every task statechange (
RUNNINGat start,FINISHED/FAILED/KILLEDat end) -- roughly two or more per task,on the driver's RPC intake path. This makes it
Externalizablewith a compact manual encodinginstead of relying on default Java serialization.
The manual form mirrors
UpdateBlockInfoin the same message family, and reusesTaskDescription's exact wire form for the fractional-CPUBigDecimal:state-> one byte (.id) /TaskState(id)taskCpus-> normalized decimal string viaCpuAmount.toDisplayString/CpuAmount.normalizedata->SerializableBuffer's existing channel-based serialization (no extra copy for largetask results)
resources-> a size-prefixed nested mapWhy are the changes needed?
Measuring the Java-serialized size (a fresh
ObjectOutputStreamper RPC, so no class-descriptorcaching) shows an empty-payload
StatusUpdateis 1713 bytes, dominated by two fields:state(a ScalaEnumerationvalue)taskCpus(aBigDecimal)resources(emptyMap)data(emptySerializableBuffer)StatusUpdateA Scala
Enumeration.Valueserializes a reference to its enclosingEnumerationobject, so itdrags in the whole
TaskStateobject;scala.math.BigDecimaldrags injava.math.BigDecimal+BigInteger+MathContext+RoundingModedescriptors. The actual payload is ~31 bytes.The manual encoding shrinks the empty-payload message from 1713 bytes to 191 bytes (~9x
smaller), cutting steady serialization/GC/bandwidth on a per-task-frequency control message. This
reduces allocation churn on the driver's RPC intake path; it is not a throughput change and no
benchmark claim is made.
Does this PR introduce any user-facing change?
No.
StatusUpdateis an internal driver<->executor RPC message. All fields round-trip identically(fractional CPUs exactly, via the same
CpuAmountformTaskDescriptionalready uses), and Javaserialization of these messages is already documented as not stable across Spark versions.
How was this patch tested?
New
CoarseGrainedClusterMessagesSuiteround-tripsStatusUpdatethroughJavaSerializer(theRPC serializer): all fields with a non-empty payload and nested
resources, everyTaskState,fractional
taskCpusexactly (value and scale), an empty payload, and a size guard confirming themessage is now well under 512 bytes.
Was this patch authored or co-authored using generative AI tooling?
Generated-by: Isaac
This pull request and its description were written by Isaac.