Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/10835#discussion_r50768513
--- Diff: core/src/main/scala/org/apache/spark/executor/InputMetrics.scala
---
@@ -34,44 +35,73 @@ object DataReadMethod extends Enumeration with
Serializable {
/**
* :: DeveloperApi ::
- * Metrics about reading input data.
+ * A collection of accumulators that represents metrics about reading data
from external systems.
*/
@DeveloperApi
-case class InputMetrics(readMethod: DataReadMethod.Value) {
+class InputMetrics private (
+ _bytesRead: Accumulator[Long],
+ _recordsRead: Accumulator[Long],
+ _readMethod: Accumulator[String])
+ extends Serializable {
+
+ private[executor] def this(accumMap: Map[String, Accumulator[_]]) {
+ this(
+ TaskMetrics.getAccum[Long](accumMap,
InternalAccumulator.input.BYTES_READ),
+ TaskMetrics.getAccum[Long](accumMap,
InternalAccumulator.input.RECORDS_READ),
+ TaskMetrics.getAccum[String](accumMap,
InternalAccumulator.input.READ_METHOD))
+ }
/**
- * This is volatile so that it is visible to the updater thread.
+ * Create a new [[InputMetrics]] that is not associated with any
particular task.
+ *
+ * This mainly exists because of SPARK-5225, where we are forced to use
a dummy [[InputMetrics]]
+ * because we want to ignore metrics from a second read method. In the
future, we should revisit
+ * whether this is needed.
+ *
+ * A better alternative is [[TaskMetrics.registerInputMetrics]].
*/
- @volatile @transient var bytesReadCallback: Option[() => Long] = None
+ private[executor] def this() {
+ this(InternalAccumulator.createInputAccums()
+ .map { a => (a.name.get, a) }.toMap[String, Accumulator[_]])
+ }
/**
- * Total bytes read.
+ * Total number of bytes read.
*/
- private var _bytesRead: Long = _
- def bytesRead: Long = _bytesRead
- def incBytesRead(bytes: Long): Unit = _bytesRead += bytes
+ def bytesRead: Long = _bytesRead.localValue
/**
- * Total records read.
+ * Total number of records read.
*/
- private var _recordsRead: Long = _
- def recordsRead: Long = _recordsRead
- def incRecordsRead(records: Long): Unit = _recordsRead += records
+ def recordsRead: Long = _recordsRead.localValue
/**
- * Invoke the bytesReadCallback and mutate bytesRead.
+ * The source from which this task reads its input.
*/
- def updateBytesRead() {
- bytesReadCallback.foreach { c =>
- _bytesRead = c()
- }
+ def readMethod: DataReadMethod.Value =
DataReadMethod.withName(_readMethod.localValue)
+
+ private[spark] def incBytesRead(v: Long): Unit = _bytesRead.add(v)
--- End diff --
Yeah, I can't imagine that anyone would have reason to call
`setBytesCallback`, so this is fine.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
