You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "Yohahaha (via GitHub)" <gi...@apache.org> on 2025/11/14 07:37:22 UTC
[PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Yohahaha opened a new pull request, #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083
<!--
Thank you for submitting a pull request! Here are some tips:
1. For first-time contributors, please read our contributing guide:
https://github.com/apache/incubator-gluten/blob/main/CONTRIBUTING.md
2. If necessary, create a GitHub issue for discussion beforehand to avoid duplicate work.
3. If the PR is specific to a single backend, include [VL] or [CH] in the PR title to indicate the
Velox or ClickHouse backend, respectively.
4. If the PR is not ready for review, please mark it as a draft.
-->
## What changes are proposed in this pull request?
<!--
Provide a clear and concise description of the changes introduced in this PR.
Ensure the PR description aligns with the code changes, especially after updates.
If applicable, include "Fixes #<GitHub_Issue_ID>" to automatically close the corresponding issue
when the PR is merged.
-->
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?
<!--
Describe how the changes were tested, if applicable.
Include new tests to validate the functionality, if necessary.
For UI-related changes, attach screenshots to demonstrate the updates.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on code in PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#discussion_r2546001167
##########
backends-velox/src/main/scala/org/apache/spark/shuffle/ColumnarShuffleWriter.scala:
##########
@@ -203,7 +205,7 @@ class ColumnarShuffleWriter[K, V](
val columnarBatchHandle =
ColumnarBatches.getNativeHandle(BackendsApiManager.getBackendName, cb)
val startTime = System.nanoTime()
- shuffleWriterJniWrapper.write(
+ val curBytesWritten = shuffleWriterJniWrapper.write(
Review Comment:
`Java_org_apache_gluten_vectorized_ShuffleWriterJniWrapper_write` I mean the result of this method was not used before, after this pr, its results can be used to update shuffle written bytes per batch
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#issuecomment-3561098281
@marin-ma thank you for the detailed review, it's my mistake for lost some codes when rebasing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#discussion_r2537229964
##########
backends-velox/src/main/scala/org/apache/spark/shuffle/ColumnarShuffleWriter.scala:
##########
@@ -203,7 +205,7 @@ class ColumnarShuffleWriter[K, V](
val columnarBatchHandle =
ColumnarBatches.getNativeHandle(BackendsApiManager.getBackendName, cb)
val startTime = System.nanoTime()
- shuffleWriterJniWrapper.write(
+ val curBytesWritten = shuffleWriterJniWrapper.write(
Review Comment:
https://github.com/apache/incubator-gluten/blob/main/cpp/core/jni/JniWrapper.cc#L1000-L1002
Looks like the returned value from each call is the size of the input batch, but looks like here it's treated as an accumulated value. I wonder if this should be
```
val bytesWritten = shuffleWriterJniWrapper.write(...)
writeMetrics.incBytesWritten(bytesWritten)
```
##########
cpp/core/shuffle/PartitionWriter.h:
##########
@@ -48,7 +48,12 @@ class PartitionWriter : public Reclaimable {
~PartitionWriter() override = default;
- virtual arrow::Status stop(ShuffleWriterMetrics* metrics) = 0;
+ void acceptMetrics(ShuffleWriterMetrics* metrics) {
Review Comment:
Why do we need this change?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on code in PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#discussion_r2540189989
##########
cpp/core/shuffle/PartitionWriter.h:
##########
@@ -48,7 +48,12 @@ class PartitionWriter : public Reclaimable {
~PartitionWriter() override = default;
- virtual arrow::Status stop(ShuffleWriterMetrics* metrics) = 0;
+ void acceptMetrics(ShuffleWriterMetrics* metrics) {
Review Comment:
use metrics member to count written bytes per batch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#issuecomment-3594390675
@marin-ma please help review again, thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on code in PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#discussion_r2579422668
##########
cpp/core/shuffle/ShuffleWriter.cc:
##########
@@ -61,8 +61,8 @@ int32_t ShuffleWriter::numPartitions() const {
return numPartitions_;
}
-int64_t ShuffleWriter::totalBytesWritten() const {
- return metrics_.totalBytesWritten;
+int64_t ShuffleWriter::bytesWritten() const {
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on code in PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#discussion_r2579434438
##########
backends-velox/src-celeborn/main/scala/org/apache/spark/shuffle/VeloxCelebornColumnarShuffleWriter.scala:
##########
@@ -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(
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on code in PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#discussion_r2544070346
##########
backends-velox/src/main/scala/org/apache/spark/shuffle/ColumnarShuffleWriter.scala:
##########
@@ -203,7 +205,7 @@ class ColumnarShuffleWriter[K, V](
val columnarBatchHandle =
ColumnarBatches.getNativeHandle(BackendsApiManager.getBackendName, cb)
val startTime = System.nanoTime()
- shuffleWriterJniWrapper.write(
+ val curBytesWritten = shuffleWriterJniWrapper.write(
Review Comment:
after this pr, `Java_org_apache_gluten_vectorized_ShuffleWriterJniWrapper_write` will return the bytes we written to shuffle service during this call which was compressed and accurate.
`batch->numBytes` is not accurate right? and we did not use it in java side before, so this pr wants to use the correct return value to enhance observability.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on code in PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#discussion_r2540187133
##########
backends-velox/src/main/scala/org/apache/spark/shuffle/ColumnarShuffleWriter.scala:
##########
@@ -203,7 +205,7 @@ class ColumnarShuffleWriter[K, V](
val columnarBatchHandle =
ColumnarBatches.getNativeHandle(BackendsApiManager.getBackendName, cb)
val startTime = System.nanoTime()
- shuffleWriterJniWrapper.write(
+ val curBytesWritten = shuffleWriterJniWrapper.write(
Review Comment:
shuffle write bytes does not equals to batch size, so before logic was wrong.
> I wonder if this should be
it depends on where we count incremental bytes, currently jni call returns total written bytes but spark only expose incremental interface.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#issuecomment-3605085442
@Yohahaha Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#issuecomment-3562466294
@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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#issuecomment-3569010670
> @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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#discussion_r2576006246
##########
backends-velox/src-celeborn/main/scala/org/apache/spark/shuffle/VeloxCelebornColumnarShuffleWriter.scala:
##########
@@ -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(
Review Comment:
I think `VeloxUniffleColumnarShuffleWriter` also needs to be updated. It's using the same native partition writer as celeborn.
##########
backends-velox/src/main/scala/org/apache/spark/shuffle/ColumnarShuffleWriter.scala:
##########
@@ -223,6 +223,7 @@ class ColumnarShuffleWriter[K, V](
dep.metrics("numInputRows").add(rows)
dep.metrics("inputBatches").add(1)
// This metric is important, AQE use it to decide if EliminateLimit
+ writeMetrics.incBytesWritten(bytesWritten)
Review Comment:
Please move the comment above this line to below.
##########
cpp/core/shuffle/ShuffleWriter.cc:
##########
@@ -61,8 +61,8 @@ int32_t ShuffleWriter::numPartitions() const {
return numPartitions_;
}
-int64_t ShuffleWriter::totalBytesWritten() const {
- return metrics_.totalBytesWritten;
+int64_t ShuffleWriter::bytesWritten() const {
Review Comment:
Can we simply modify the signature of`ShuffleWriter::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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#discussion_r2541180332
##########
backends-velox/src/main/scala/org/apache/spark/shuffle/ColumnarShuffleWriter.scala:
##########
@@ -203,7 +205,7 @@ class ColumnarShuffleWriter[K, V](
val columnarBatchHandle =
ColumnarBatches.getNativeHandle(BackendsApiManager.getBackendName, cb)
val startTime = System.nanoTime()
- shuffleWriterJniWrapper.write(
+ val curBytesWritten = shuffleWriterJniWrapper.write(
Review Comment:
```
JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_ShuffleWriterJniWrapper_write( // NOLINT
JNIEnv* env,
jobject wrapper,
jlong shuffleWriterHandle,
jint numRows,
jlong batchHandle,
jlong memLimit) {
JNI_METHOD_START
auto shuffleWriter = ObjectStore::retrieve<ShuffleWriter>(shuffleWriterHandle);
if (!shuffleWriter) {
std::string errorMessage = "Invalid shuffle writer handle " + std::to_string(shuffleWriterHandle);
throw GlutenException(errorMessage);
}
// The column batch maybe VeloxColumnBatch or ArrowCStructColumnarBatch(FallbackRangeShuffleWriter)
auto batch = ObjectStore::retrieve<ColumnarBatch>(batchHandle);
auto numBytes = batch->numBytes();
arrowAssertOkOrThrow(shuffleWriter->write(batch, memLimit), "Native write: shuffle writer failed");
return numBytes;
JNI_METHOD_END(kInvalidObjectHandle)
}
```
Seems like it only returns one input batch size. Am I misunderstanding something?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#issuecomment-3546022693
@kerwin-zk @marin-ma please help review this pr, thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083#discussion_r2545468458
##########
backends-velox/src/main/scala/org/apache/spark/shuffle/ColumnarShuffleWriter.scala:
##########
@@ -203,7 +205,7 @@ class ColumnarShuffleWriter[K, V](
val columnarBatchHandle =
ColumnarBatches.getNativeHandle(BackendsApiManager.getBackendName, cb)
val startTime = System.nanoTime()
- shuffleWriterJniWrapper.write(
+ val curBytesWritten = shuffleWriterJniWrapper.write(
Review Comment:
@Yohahaha I don't see any modification to the `Java_org_apache_gluten_vectorized_ShuffleWriterJniWrapper_write` in this patch.
> `batch->numBytes` is not accurate right?
I'm also not aware of that. Could you elaborate in which cases it could be inaccurate? If the value is wrong we should fix this api as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Update shuffle write metrics per batch [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma merged PR #11083:
URL: https://github.com/apache/incubator-gluten/pull/11083
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org