[CELEBORN-2439][TEZ] Report accurate partition statistics for shuffle output - #3821
[CELEBORN-2439][TEZ] Report accurate partition statistics for shuffle output#3821SparksFyz wants to merge 1 commit into
Conversation
| import java.util.concurrent.atomic.LongAdder; | ||
|
|
||
| import org.junit.Test; | ||
| import org.mockito.InOrder; |
There was a problem hiding this comment.
Could we add focused coverage for the two Tez consumers changed in this patch? This suite verifies the counters inside CelebornTezWriter, but it would still pass if CelebornSorter or CelebornUnorderedPartitionedKVWriter failed to expose the snapshot, or if disabled partition-stat reporting returned a non-null array. Exercising enabled and disabled TEZ_RUNTIME_REPORT_PARTITION_STATS through both paths would protect the user-visible behavior this PR is fixing.
There was a problem hiding this comment.
@SteNicholas Thanks for the review. I added focused tests for both CelebornSorter and CelebornUnorderedPartitionedKVWriter, covering explicit PRECISE, the default MEMORY_OPTIMIZED, and disabled (NONE) partition-stat reporting. The tests verify that enabled reporting exposes the snapshot after CelebornTezWriter is closed, while disabled reporting returns null without querying the writer.
43306de to
7f0c877
Compare
What changes were proposed in this pull request?
This pull request reports accurate per-partition size statistics for Celeborn Tez shuffle output.
The main changes are:
DataPusherinCelebornTezWriter.mergeData.bytesPerPartitioncounters fromCelebornSortBasedPusher.Why are the changes needed?
The current Tez shuffle implementation reports incorrect partition statistics:
partitionStatsarray is never populated.bytesPerPartitionis incremented once per record.These incorrect statistics may affect Tez partition-size estimation and reducer parallelism decisions.
Does this PR resolve a correctness bug?
Does this PR introduce any user-facing change?
How was this patch tested?
Added
CelebornTezWriterSuiteJto verify:mergeDataaccounting using the bytes returned byShuffleClient.pushDataaccounting after the writer is closed.DataPusher.waitOnTermination,pushMergedData, andmapperEnd.