[VL] Update shuffle write metrics per batch#11083
Conversation
47d339c to
af3860e
Compare
|
@kerwin-zk @marin-ma please help review this pr, thank you! |
|
@marin-ma thank you for the detailed review, it's my mistake for lost some codes when rebasing. |
|
@Yohahaha Thanks for the clarification and updates. Can we simply add an API for the PartitionWriter to count the total written bytes, and calculate then return the delta of the bytes written in the native ShuffleWriter? |
thank you for the suggestion, I will refactor it later. |
f673f41 to
6dbd27e
Compare
|
@marin-ma please help review again, thank you! |
| @@ -77,7 +77,7 @@ class VeloxCelebornColumnarShuffleWriter[K, V]( | |||
| val columnarBatchHandle = | |||
| ColumnarBatches.getNativeHandle(BackendsApiManager.getBackendName, cb) | |||
| val startTime = System.nanoTime() | |||
| shuffleWriterJniWrapper.write( | |||
| val bytesWritten = shuffleWriterJniWrapper.write( | |||
There was a problem hiding this comment.
I think VeloxUniffleColumnarShuffleWriter also needs to be updated. It's using the same native partition writer as celeborn.
There was a problem hiding this comment.
Uniffle does not call incBytesWritten and the logic of metrics collecting differs with local and celeborn, it's better to let related developer to change it.
| @@ -61,8 +61,8 @@ int32_t ShuffleWriter::numPartitions() const { | |||
| return numPartitions_; | |||
| } | |||
|
|
|||
| int64_t ShuffleWriter::totalBytesWritten() const { | |||
| return metrics_.totalBytesWritten; | |||
| int64_t ShuffleWriter::bytesWritten() const { | |||
There was a problem hiding this comment.
Can we simply modify the signature ofShuffleWriter::write and ShuffleWriter::stop to return the bytes evicted as Result<int64_t>? And seems like this value is not updated and used for local shuffle write, please add some explanation that only rss shuffle write returns the evicted bytes.
There was a problem hiding this comment.
Can we simply modify the signature ofShuffleWriter::write and ShuffleWriter::stop to return the bytes evicted as Result<int64_t>
The write api can be easily modified, but the stop api is not, stop api collect and return whole shuffle metrics of task, I prefer do refactor work for api change in another pr and make current pr as simple as possible.
98784f0 to
e1bee6d
Compare
|
@marin-ma any more comments? |
|
@Yohahaha Thanks |
3 participants
What changes are proposed in this pull request?
In our prod env, some stages write out hundreds of terabytes of shuffle data, but current UI does not shown it.
How was this patch tested?