wanghaihua created SPARK-58830:
----------------------------------
Summary: Shuffle can silently drop data when MapStatus metadata is
corrupted between mapper and reducer
Key: SPARK-58830
URL: https://issues.apache.org/jira/browse/SPARK-58830
Project: Spark
Issue Type: Improvement
Components: Spark Core
Affects Versions: 3.4.4
Reporter: wanghaihua
Attachments: shuffle-datalost.png
h3. Problem
{{MapStatus}} is one of the most frequently accessed metadata structures
throughout the shuffle lifecycle: every mapper produces one, the driver retains
one for each map task, and every reducer consults it to determine which shuffle
blocks to fetch.
If a {{MapStatus}} instance—specifically its {{partitionLengths}} or compressed
representation—is corrupted in memory or in transit between the mapper and
reducer, a partition containing data on disk may incorrectly appear to have a
size of zero. Possible causes include transient hardware faults, JVM or memory
corruption, and serialization bugs.
Shuffle block fetchers intentionally skip zero-length blocks when:
{{MapStatus.getSizeForBlock(reduceId) == 0}}
Consequently, the reducer silently skips the affected block. The valid on-disk
shuffle data is never read and is effectively lost, with no exception, failed
stage, or log signal anywhere in the pipeline.
We observed this behavior in production. A job produced fewer output rows than
an otherwise equivalent rerun, despite there being no errors in the driver or
executor logs. Comparing the Spark UI SQL metrics from the two runs revealed a
significant discrepancy between shuffle “records written” and “records read” at
an {{Exchange}} node in the affected run.
After instrumenting {{MapStatus}} at multiple points along the mapper → driver
→ reducer path, we found that several non-empty partition entries had changed
to zero, while the corresponding shuffle data files on disk remained complete
and correct. The root cause was traced to a hardware fault that silently
corrupted a {{MapStatus}} object held in the driver’s memory.
h3. Related Existing Safeguards
* *SPARK-35276* added checksums for shuffle data files, allowing Spark to
detect disk- or I/O-level corruption when a reducer reads the shuffle bytes.
However, when the corruption affects metadata rather than data—for example,
when {{MapStatus}} incorrectly reports a non-empty partition as empty—the
reducer never attempts to read the block, so the data-checksum path is not
exercised.
* *SPARK-40872* handles a related but distinct push-based shuffle scenario in
which a merged chunk is empty because it was produced by a faulty merge node.
It recovers by falling back to the original blocks, but it does not cover
non-empty-to-empty corruption in a mapper’s own {{MapStatus}} outside
push-based merging.
* *SPARK-57491* addresses a scenario where two attempts for the same
partition—such as speculative attempts with non-deterministic shuffle keys—both
complete their pushes, but only one attempt’s {{MapStatus}} survives. It does
not address in-memory or in-transit corruption of an individual
{{{}MapStatus{}}}.
None of these safeguards detects a {{MapStatus}} that was correct when produced
by the mapper but became corrupted before being consumed by a reducer.
h3. Proposed Solution
Add an integrity checksum to {{{}MapStatus{}}}, computed over the {*}set of
non-empty partition indices{*}, rather than over the partition sizes themselves.
Partition sizes are already represented using lossy compression or estimation
in {{CompressedMapStatus}} and {{{}HighlyCompressedMapStatus{}}}. Their exact
encoded values may therefore legitimately differ while remaining correct. For
this failure mode, the important invariant is whether each partition is empty
or non-empty.
The checksum would travel with {{MapStatus}} throughout its normal lifecycle,
including Java serialization, and would be verified whenever the metadata is
consumed to plan shuffle block fetches.
A checksum mismatch would indicate that the metadata was corrupted in transit
or at rest. Spark could then throw a {{{}FetchFailedException{}}}, reusing the
existing fetch-failure and stage-retry recovery path instead of silently
producing incomplete results.
h3. Design Notes
* Store the checksum as an {{{}Option[Int]{}}}, gated by a new internal
configuration that is disabled by default, allowing the feature to be enabled
selectively.
* {{None}} naturally represents both “checksum verification disabled” and “no
checksum present,” without reserving a sentinel value in the 32-bit checksum
space. A computed checksum remains valid even when there are no non-empty
partitions.
* Make the checksum algorithm configurable, initially supporting {{CRC32}} and
{{{}Adler32{}}}.
* Perform verification where {{MapStatus}} is consumed to plan block fetches.
* On mismatch, throw {{FetchFailedException}} and reuse Spark’s existing
map-output fetch-failure handling rather than introducing a new recovery
mechanism.
* Verification requires {{O(N)}} time per {{{}MapStatus{}}}, where {{N}} is
the number of reduce partitions, plus a small constant number of bytes per
{{MapStatus}} for the optional checksum.
We have a working prototype that includes:
* A new {{MapStatusChecksum}} utility.
* A {{nonEmptyChecksum: Option[Int]}} field on {{{}MapStatus{}}}.
* Serialization support for the checksum.
* Checksum verification integrated into {{{}MapOutputTracker{}}}.
We would like to contribute this implementation upstream.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
