[
https://issues.apache.org/jira/browse/SPARK-58830?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
wanghaihua updated SPARK-58830:
-------------------------------
Description:
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.
!shuffle-datalost.png|width=685,height=397!
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.
was:
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.
> 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.
> !shuffle-datalost.png|width=685,height=397!
> 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]
