[
https://issues.apache.org/jira/browse/SPARK-58830?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
wanghaihua updated SPARK-58830:
-------------------------------
Attachment: shuffle-datalost.png
> 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
> Priority: Major
> 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]
