[SPARK-20703][SQL] Associate metrics with data writes onto DataFrameWriter operations #18159
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,108 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.spark.sql.execution.command | ||
|
|
||
| import org.apache.spark.SparkContext | ||
| import org.apache.spark.sql.{Row, SparkSession} | ||
| import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow} | ||
| import org.apache.spark.sql.catalyst.plans.logical | ||
| import org.apache.spark.sql.execution.{SparkPlan, SQLExecution} | ||
| import org.apache.spark.sql.execution.datasources.ExecutedWriteSummary | ||
| import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics} | ||
| import org.apache.spark.util.Utils | ||
|
|
||
| /** | ||
| * A logical command specialized for writing data out. `FileWritingCommand`s are | ||
| * wrapped in `FileWritingCommandExec` during execution. | ||
| */ | ||
| trait FileWritingCommand extends logical.Command { | ||
|
|
||
| // The caller of `FileWritingCommand` can replace the metrics location by providing this external | ||
| // metrics structure. | ||
| private var _externalMetrics: Option[Map[String, SQLMetric]] = None | ||
| private[sql] def withExternalMetrics(map: Map[String, SQLMetric]): this.type = { | ||
| _externalMetrics = Option(map) | ||
| this | ||
| } | ||
|
Contributor
There was a problem hiding this comment. It's not clear to me why we need this and the comment is not helpful.
Member
Author
There was a problem hiding this comment. I've explained the purpose of this in below comments. Once a command invokes another command to write data. We have to update the metrics in the physical plan of the first command, not the second. Before we invoke the second command, we have to replace the metrics in the physical plan's metrics with the metrics in the first one (i.e., the external metrics). It seems to me that you miss the relationship that there is two different commands.
Contributor
There was a problem hiding this comment. Indeed, I missed it at first, but then I got it. See below: #18159 (comment)
Member
Author
There was a problem hiding this comment. I should have better code comment on this part too. I'll add it in next commit. |
||
|
|
||
| /** | ||
| * Those metrics will be updated once the command finishes writing data out. Those metrics will | ||
| * be taken by `FileWritingCommandExec` as its metrics when showing in UI. | ||
| */ | ||
| def metrics(sparkContext: SparkContext): Map[String, SQLMetric] = _externalMetrics.getOrElse { | ||
| Map( | ||
| // General metrics. | ||
| "avgTime" -> SQLMetrics.createMetric(sparkContext, "average writing time (ms)"), | ||
| "numFiles" -> SQLMetrics.createMetric(sparkContext, "number of written files"), | ||
| "numOutputBytes" -> SQLMetrics.createMetric(sparkContext, "bytes of written output"), | ||
| "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output rows"), | ||
| "numParts" -> SQLMetrics.createMetric(sparkContext, "number of dynamic part") | ||
| ) | ||
| } | ||
|
|
||
| /** | ||
| * Callback function that update metrics collected from the writing operation. | ||
| */ | ||
| private[sql] def postDriverMetrics(sparkContext: SparkContext, metrics: Map[String, SQLMetric]) | ||
| (writeSummaries: Seq[ExecutedWriteSummary]): Unit = { | ||
| var numPartitions = 0 | ||
| var numFiles = 0 | ||
| var totalNumBytes: Long = 0L | ||
| var totalNumOutput: Long = 0L | ||
|
|
||
| writeSummaries.foreach { summary => | ||
| numPartitions += summary.updatedPartitions.size | ||
| numFiles += summary.numOutputFile | ||
| totalNumBytes += summary.numOutputBytes | ||
| totalNumOutput += summary.numOutputRows | ||
| } | ||
|
|
||
| // The time for writing individual file can be zero if it's less than 1 ms. Zero values can | ||
| // lower actual time of writing when calculating average, so excluding them. | ||
|
Contributor
There was a problem hiding this comment. Seems arbitrary to exclude 0 values. 1ms values also bring the average down, yet you're keeping those.
Member
Author
There was a problem hiding this comment. Excluding 0 values because 0 values can make the average writing time as 0. It seems to me that it doesn't really make sense to show 0 writing time. Seems it indicates the writing doesn't cost time. 1ms brings the average down, but you still get a meaningful time value. |
||
| val writingTime = | ||
| Utils.average(writeSummaries.flatMap(_.writingTimePerFile.filter(_ > 0))).toLong | ||
|
|
||
| val metricsNames = metrics.keys.toSeq.sorted | ||
|
Contributor
There was a problem hiding this comment. how do you guarantee the
Member
Author
There was a problem hiding this comment.
|
||
| val metricsValues = Seq(writingTime, numFiles, totalNumBytes, totalNumOutput, numPartitions) | ||
| metricsNames.zip(metricsValues).foreach(x => metrics(x._1).add(x._2)) | ||
|
Contributor
There was a problem hiding this comment. This doesn't make sense if
Member
Author
There was a problem hiding this comment. I am not sure if I correctly understand your point. The external metrics are always coming from the same kind of trait. It's not coming from user, how it's user-provided map? The order of keys returned by
Contributor
There was a problem hiding this comment. I meant user as in caller of this function. This function only works when the input satisfies some requirements and silently fails otherwise. I get it that it's a private method, but it's very error prone. Imagine I want to extend this by adding another metric called At the very least, leave a comment saying that the
Member
Author
There was a problem hiding this comment. For metrics purpose, this change is a bit too large. I'd try not to increase the complexity for now. I added a comment for this. |
||
|
|
||
| val executionId = sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY) | ||
| SQLMetrics.postDriverMetricUpdates(sparkContext, executionId, metricsNames.map(metrics(_))) | ||
| } | ||
|
|
||
| def run( | ||
| sparkSession: SparkSession, | ||
| children: Seq[SparkPlan], | ||
| metrics: Map[String, SQLMetric], | ||
| metricsCallback: (Seq[ExecutedWriteSummary]) => Unit): Seq[Row] | ||
|
Contributor
There was a problem hiding this comment. Why does this take A big part of this patch is just about passing down metrics through this interface.. It also makes it quite hard to follow. Is there no way we can avoid this? It would be a significant improvement if you could find a solution. Basically try to specify the metrics (either the Map you have, or None) together with the corresponding callback function just once, when you first instantiate
Member
Author
There was a problem hiding this comment. The case is more complicated... We have those commands which don't write the data but invoke another commands to do that. The execution data for showing on UI is bound to the original commands, but the invoked commands. The invoked commands will update the metrics in its physical plans, not the caller's. If we don't pass the metrics or callback function, you won't actually update the correct metrics. That's why we need to pass metrics or callback function here...
Contributor
There was a problem hiding this comment. I understand that, but I think it can still be simplified. See below: #18159 (comment) |
||
| } | ||
|
|
||
| /** | ||
| * A physical operator specialized to execute the run method of a `FileWritingCommand`, | ||
| * save the result to prevent multiple executions, and record necessary metrics for UI. | ||
| */ | ||
| case class FileWritingCommandExec( | ||
| cmd: FileWritingCommand, | ||
| children: Seq[SparkPlan]) extends CommandExec { | ||
|
|
||
| override lazy val metrics = cmd.metrics(sqlContext.sparkContext) | ||
|
|
||
| protected[sql] lazy val invokeCommand: Seq[Row] = | ||
| cmd.run(sqlContext.sparkSession, children, metrics, | ||
| cmd.postDriverMetrics(sqlContext.sparkContext, metrics)) | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
shall we require array type parameter? We are calling
sizetwice in this method, which can be very slow if the input is not indexed seq.