Skip to content

Commit

Permalink
[CELEBORN-1265] Fix batches read metric for gluten columnar shuffle
Browse files Browse the repository at this point in the history
### What changes were proposed in this pull request?
Fix batches read metric for gluten columnar shuffle

### Why are the changes needed?
![image](https://github.com/apache/incubator-celeborn/assets/107825064/c862e83b-8e3e-4705-a151-41e5b6675d7a)

Due to the fix in [Gluten-4025](apache/incubator-gluten#4051) for the records read metric issue, the read metric of CelebornShuffleReader does not need additional processing, otherwise the batches read metric will have the issue shown in the graph.

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
CI

Closes #2289 from kerwin-zk/batches-read.

Authored-by: xiyu.zk <xiyu.zk@alibaba-inc.com>
Signed-off-by: zky.zhoukeyong <zky.zhoukeyong@alibaba-inc.com>
  • Loading branch information
kerwin-zk authored and waitinfuture committed Feb 7, 2024
1 parent ab4c0bc commit 7b2b9f7
Show file tree
Hide file tree
Showing 2 changed files with 4 additions and 16 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -173,14 +173,11 @@ class CelebornShuffleReader[K, C](
}

val iterWithUpdatedRecordsRead =
if (GlutenShuffleDependencyHelper.isGlutenDep(dep.getClass.getName)) {
GlutenShuffleDependencyHelper.withUpdatedRecordsRead(recordIter, metrics)
} else {
recordIter.map { record =>
metrics.incRecordsRead(1)
record
}
recordIter.map { record =>
metrics.incRecordsRead(1)
record
}

val metricIter = CompletionIterator[(Any, Any), Iterator[(Any, Any)]](
iterWithUpdatedRecordsRead,
context.taskMetrics().mergeShuffleReadMetrics())
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,13 +32,4 @@ object GlutenShuffleDependencyHelper {
// scalastyle:on
"org.apache.spark.shuffle.ColumnarShuffleDependency".equals(depName)
}

def withUpdatedRecordsRead(
input: Iterator[(Any, Any)],
metrics: ShuffleReadMetricsReporter): Iterator[(Any, Any)] = {
input.map { record =>
metrics.incRecordsRead(record._2.asInstanceOf[ColumnarBatch].numRows())
record
}
}
}

0 comments on commit 7b2b9f7

Please sign in to comment.