peter-toth commented on code in PR #58591:
URL: https://github.com/apache/spark/pull/58591#discussion_r3956120700
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -274,6 +307,34 @@ case class GroupPartitionsExec(
@transient private lazy val hasCoalescing: Boolean =
groupedPartitions.exists(_._2.size > 1)
+ // All values are computed on the driver by `grouping`, so they are reported
through
+ // `sendDriverMetrics` rather than task-side accumulators. Registration
reads constructor
+ // parameters only: pruning and replication are facts of the alignment path
and are registered
+ // only in the modes that produce them, while coalescing can happen in any
mode and reports 0
+ // when nothing merged.
+ @transient override lazy val metrics: Map[String, SQLMetric] = Map(
+ "numInputPartitions" -> SQLMetrics.createMetric(sparkContext, "number of
input partitions"),
+ "numPartitions" -> SQLMetrics.createMetric(sparkContext, "number of
partitions"),
+ "numEmptyPartitions" -> SQLMetrics.createMetric(sparkContext, "number of
empty partitions"),
+ "numCoalescedPartitions" ->
+ SQLMetrics.createMetric(sparkContext, "number of coalesced partitions"),
+ "maxPartitionsPerGroup" ->
+ SQLMetrics.createMetric(sparkContext, "max partitions per group")) ++ {
+ if (expectedPartitionKeys.isDefined && !distributePartitions) {
+ Map("numReplicatedPartitions" -> SQLMetrics.createMetric(sparkContext,
+ "number of replicated input partition reads"))
Review Comment:
**Finding 1.** Same diagnosis here, reached independently. The suggested
condition needs a matching change at `GroupPartitionsExec.scala:452` though, or
it turns a cosmetic problem into a crash.
`sendDriverMetrics` repeats the old predicate to decide what to set:
```scala
if (expectedPartitionKeys.isDefined && !distributePartitions) {
set("numReplicatedPartitions", grouping.numReplicatedPartitions)
}
```
On a plain SPJ the metric is then no longer registered but is still set, and
`metrics(name)` throws. I applied the suggestion at the registration site only,
on `358ca646ade`, and `GroupPartitionsExecSuite` goes from 29 passing to 27:
```
- SPARK-59310: an inner join intersection prunes both sides *** FAILED ***
java.util.NoSuchElementException: key not found: numReplicatedPartitions
- SPARK-59310: a disjoint inner join prunes both sides to empty end to end
*** FAILED ***
java.util.NoSuchElementException: key not found: numReplicatedPartitions
```
That is every plain storage-partitioned join, thrown from `doExecute`.
The two predicates sit 130 lines apart and can only agree by inspection, so
removing the second copy looks better than editing it. Letting `set` skip an
unregistered name leaves one source of truth:
```scala
def set(name: String, value: Long): Unit = metrics.get(name).foreach {
metric =>
metric.set(value)
driverAccumUpdates += (metric.id -> value)
}
```
with both `if`s in `sendDriverMetrics` becoming plain `set` calls. It trades
a loud failure for a silent one on a mistyped name; keying the conditionals off
`metrics.contains(...)` is the same idea without that trade.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -359,7 +420,47 @@ case class GroupPartitionsExec(
private[v2] def kWayMergeOrdering: Seq[SortOrder] =
child.outputOrdering.map(_.copy(sameOrderExpressions = Seq.empty))
+ private def sendDriverMetrics(): Unit = {
+ val executionId =
sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
+ val driverAccumUpdates = ArrayBuffer.empty[(Long, Long)]
+ def set(name: String, value: Long): Unit = {
+ val metric = metrics(name)
+ metric.set(value)
+ driverAccumUpdates += (metric.id -> value)
+ }
+ // A single pass for the three per-group counts; an empty group and a
coalesced one are
+ // mutually exclusive.
+ var numEmptyPartitions = 0
+ var numCoalescedPartitions = 0
+ var maxPartitionsPerGroup = 0
+ groupedPartitions.foreach { case (_, group) =>
+ val size = group.size
+ if (size == 0) {
+ numEmptyPartitions += 1
+ } else if (size > 1) {
+ numCoalescedPartitions += 1
+ }
+ if (size > maxPartitionsPerGroup) {
+ maxPartitionsPerGroup = size
+ }
+ }
+ set("numInputPartitions", grouping.numInputPartitions)
+ set("numPartitions", groupedPartitions.size)
+ set("numEmptyPartitions", numEmptyPartitions)
+ set("numCoalescedPartitions", numCoalescedPartitions)
+ set("maxPartitionsPerGroup", maxPartitionsPerGroup)
+ if (expectedPartitionKeys.isDefined && !distributePartitions) {
+ set("numReplicatedPartitions", grouping.numReplicatedPartitions)
+ }
+ if (expectedPartitionKeys.isDefined) {
+ set("numPrunedPartitions", grouping.numPrunedPartitions)
+ }
+ SQLMetrics.postDriverMetricsUpdatedByValue(
+ sparkContext, executionId, driverAccumUpdates.toSeq)
+ }
+
override protected def doExecute(): RDD[InternalRow] = {
+ sendDriverMetrics()
Review Comment:
The append semantics are right: `driverAccumUpdates` is a `Seq[(Long,
Long)]` (`SQLAppStatusListener.scala:509`), `++` appends, and
`aggregateMetrics` grows the value array per id (`:278-296`), so a repeated
post really would sum under `SUM_METRIC`.
The guard already exists one level up, though. `SparkPlan.execute()` is
`executeRDD.get` over `private val executeRDD = LazyTry { doExecute() }`
(`SparkPlan.scala:186-201`), and `executeColumnar()` is the same over
`executeColumnarRDD` (`:221-236`). So `doExecute` and `doExecuteColumnar` run
at most once per plan instance whatever calls them, and a `metricsSent` lazy
val would memoize something already memoized. That is also why
`AQEShuffleReadExec` and `FileSourceScanExec` are safe -- their lazy vals are
not what protects them.
Measured on `358ca646ade`: with a `println` in `sendDriverMetrics` and two
`df.collect()` calls on one DataFrame, it fires exactly once, under the first
execution id.
That memoization has a consequence worth knowing, though it is not this PR's
to fix: the second action on a DataFrame shows a `GroupPartitions` node with no
metrics at all, because `doExecute` is not re-entered, while every task-side
metric around it is recollected. `FileSourceScanExec`'s `number of files read`
behaves identically on a second `collect()`, so it is a property of driver-side
SQL metrics generally.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
