Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-13152][CORE] Fix task metrics deprecation warning #11056

Closed
wants to merge 2 commits into from
Closed

[SPARK-13152][CORE] Fix task metrics deprecation warning #11056

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e2d518b
Make an internal non-deprecated version of incBytesRead and incRecord…
holdenk Feb 3, 2016
1076946
Add note about reason for doing this (CR feedback from Andrew Or)
holdenk Feb 3, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions core/src/main/scala/org/apache/spark/CacheManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,12 +44,12 @@ private[spark] class CacheManager(blockManager: BlockManager) extends Logging {
case Some(blockResult) =>
// Partition is already materialized, so just return its values
val existingMetrics = context.taskMetrics().registerInputMetrics(blockResult.readMethod)
existingMetrics.incBytesRead(blockResult.bytes)
existingMetrics.incBytesReadInternal(blockResult.bytes)

val iter = blockResult.data.asInstanceOf[Iterator[T]]
new InterruptibleIterator[T](context, iter) {
override def next(): T = {
existingMetrics.incRecordsRead(1)
existingMetrics.incRecordsReadInternal(1)
delegate.next()
}
}
Expand Down
4 changes: 4 additions & 0 deletions core/src/main/scala/org/apache/spark/executor/InputMetrics.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,10 +81,14 @@ class InputMetrics private (
*/
def readMethod: DataReadMethod.Value = DataReadMethod.withName(_readMethod.localValue)

// Once incBytesRead & intRecordsRead is ready to be removed from the public API
// we can remove the internal versions and make the previous public API private.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you add that this is to suppress warnings?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done :)

@deprecated("incrementing input metrics is for internal use only", "2.0.0")
def incBytesRead(v: Long): Unit = _bytesRead.add(v)
private[spark] def incBytesReadInternal(v: Long): Unit = _bytesRead.add(v)
@deprecated("incrementing input metrics is for internal use only", "2.0.0")
def incRecordsRead(v: Long): Unit = _recordsRead.add(v)
private[spark] def incRecordsReadInternal(v: Long): Unit = _recordsRead.add(v)
private[spark] def setBytesRead(v: Long): Unit = _bytesRead.setValue(v)
private[spark] def setReadMethod(v: DataReadMethod.Value): Unit =
_readMethod.setValue(v.toString)
Expand Down
4 changes: 2 additions & 2 deletions core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,7 +260,7 @@ class HadoopRDD[K, V](
finished = true
}
if (!finished) {
inputMetrics.incRecordsRead(1)
inputMetrics.incRecordsReadInternal(1)
}
if (inputMetrics.recordsRead % SparkHadoopUtil.UPDATE_INPUT_METRICS_INTERVAL_RECORDS == 0) {
updateBytesRead()
Expand Down Expand Up @@ -292,7 +292,7 @@ class HadoopRDD[K, V](
// If we can't get the bytes read from the FS stats, fall back to the split size,
// which may be inaccurate.
try {
inputMetrics.incBytesRead(split.inputSplit.value.getLength)
inputMetrics.incBytesReadInternal(split.inputSplit.value.getLength)
} catch {
case e: java.io.IOException =>
logWarning("Unable to get input size to set InputMetrics for task", e)
Expand Down
4 changes: 2 additions & 2 deletions core/src/main/scala/org/apache/spark/rdd/NewHadoopRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,7 +188,7 @@ class NewHadoopRDD[K, V](
}
havePair = false
if (!finished) {
inputMetrics.incRecordsRead(1)
inputMetrics.incRecordsReadInternal(1)
}
if (inputMetrics.recordsRead % SparkHadoopUtil.UPDATE_INPUT_METRICS_INTERVAL_RECORDS == 0) {
updateBytesRead()
Expand Down Expand Up @@ -219,7 +219,7 @@ class NewHadoopRDD[K, V](
// If we can't get the bytes read from the FS stats, fall back to the split size,
// which may be inaccurate.
try {
inputMetrics.incBytesRead(split.serializableHadoopSplit.value.getLength)
inputMetrics.incBytesReadInternal(split.serializableHadoopSplit.value.getLength)
} catch {
case e: java.io.IOException =>
logWarning("Unable to get input size to set InputMetrics for task", e)
Expand Down
4 changes: 2 additions & 2 deletions core/src/main/scala/org/apache/spark/util/JsonProtocol.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -811,8 +811,8 @@ private[spark] object JsonProtocol {
Utils.jsonOption(json \ "Input Metrics").foreach { inJson =>
val readMethod = DataReadMethod.withName((inJson \ "Data Read Method").extract[String])
val inputMetrics = metrics.registerInputMetrics(readMethod)
inputMetrics.incBytesRead((inJson \ "Bytes Read").extract[Long])
inputMetrics.incRecordsRead((inJson \ "Records Read").extractOpt[Long].getOrElse(0L))
inputMetrics.incBytesReadInternal((inJson \ "Bytes Read").extract[Long])
inputMetrics.incRecordsReadInternal((inJson \ "Records Read").extractOpt[Long].getOrElse(0L))
}

// Updated blocks
Expand Down
4 changes: 2 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SqlNewHadoopRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -214,7 +214,7 @@ private[spark] class SqlNewHadoopRDD[V: ClassTag](
}
havePair = false
if (!finished) {
inputMetrics.incRecordsRead(1)
inputMetrics.incRecordsReadInternal(1)
}
if (inputMetrics.recordsRead % SparkHadoopUtil.UPDATE_INPUT_METRICS_INTERVAL_RECORDS == 0) {
updateBytesRead()
Expand Down Expand Up @@ -246,7 +246,7 @@ private[spark] class SqlNewHadoopRDD[V: ClassTag](
// If we can't get the bytes read from the FS stats, fall back to the split size,
// which may be inaccurate.
try {
inputMetrics.incBytesRead(split.serializableHadoopSplit.value.getLength)
inputMetrics.incBytesReadInternal(split.serializableHadoopSplit.value.getLength)
} catch {
case e: java.io.IOException =>
logWarning("Unable to get input size to set InputMetrics for task", e)
Expand Down