diff options
author | Reynold Xin <rxin@cs.berkeley.edu> | 2013-04-30 13:28:35 -0700 |
---|---|---|
committer | Reynold Xin <rxin@cs.berkeley.edu> | 2013-04-30 13:28:35 -0700 |
commit | cea617457386928abfac1d04cf0cb2656fa07da0 (patch) | |
tree | 466e3468d74cca5a058900d5f8278da8e907cdce /core/src/main | |
parent | 1055785a836ab2361239f0937a1a22fee953e029 (diff) | |
parent | f708dda81ed5004325591fcc31cd79a8afa580db (diff) | |
download | spark-cea617457386928abfac1d04cf0cb2656fa07da0.tar.gz spark-cea617457386928abfac1d04cf0cb2656fa07da0.tar.bz2 spark-cea617457386928abfac1d04cf0cb2656fa07da0.zip |
Merge branch 'master' of github.com:mesos/spark into blockmanager
Conflicts:
core/src/main/scala/spark/BlockStoreShuffleFetcher.scala
Diffstat (limited to 'core/src/main')
4 files changed, 10 insertions, 60 deletions
diff --git a/core/src/main/scala/spark/BlockStoreShuffleFetcher.scala b/core/src/main/scala/spark/BlockStoreShuffleFetcher.scala index 2156efbd45..e1fb02157a 100644 --- a/core/src/main/scala/spark/BlockStoreShuffleFetcher.scala +++ b/core/src/main/scala/spark/BlockStoreShuffleFetcher.scala @@ -5,8 +5,8 @@ import scala.collection.mutable.HashMap import spark.executor.{ShuffleReadMetrics, TaskMetrics} import spark.serializer.Serializer -import spark.storage.{DelegateBlockFetchTracker, BlockManagerId} -import spark.util.{CompletionIterator, TimedIterator} +import spark.storage.BlockManagerId +import spark.util.CompletionIterator private[spark] class BlockStoreShuffleFetcher extends ShuffleFetcher with Logging { @@ -54,17 +54,16 @@ private[spark] class BlockStoreShuffleFetcher extends ShuffleFetcher with Loggin } val blockFetcherItr = blockManager.getMultiple(blocksByAddress, serializer) - val itr = new TimedIterator(blockFetcherItr.flatMap(unpackBlock)) with DelegateBlockFetchTracker - itr.setDelegate(blockFetcherItr) + val itr = blockFetcherItr.flatMap(unpackBlock) + CompletionIterator[(K,V), Iterator[(K,V)]](itr, { val shuffleMetrics = new ShuffleReadMetrics - shuffleMetrics.shuffleReadMillis = itr.getNetMillis - shuffleMetrics.remoteFetchTime = itr.remoteFetchTime - shuffleMetrics.fetchWaitTime = itr.fetchWaitTime - shuffleMetrics.remoteBytesRead = itr.remoteBytesRead - shuffleMetrics.totalBlocksFetched = itr.totalBlocks - shuffleMetrics.localBlocksFetched = itr.numLocalBlocks - shuffleMetrics.remoteBlocksFetched = itr.numRemoteBlocks + shuffleMetrics.remoteFetchTime = blockFetcherItr.remoteFetchTime + shuffleMetrics.fetchWaitTime = blockFetcherItr.fetchWaitTime + shuffleMetrics.remoteBytesRead = blockFetcherItr.remoteBytesRead + shuffleMetrics.totalBlocksFetched = blockFetcherItr.totalBlocks + shuffleMetrics.localBlocksFetched = blockFetcherItr.numLocalBlocks + shuffleMetrics.remoteBlocksFetched = blockFetcherItr.numRemoteBlocks metrics.shuffleReadMetrics = Some(shuffleMetrics) }) } diff --git a/core/src/main/scala/spark/executor/TaskMetrics.scala b/core/src/main/scala/spark/executor/TaskMetrics.scala index 93bbb6b458..a7c56c2371 100644 --- a/core/src/main/scala/spark/executor/TaskMetrics.scala +++ b/core/src/main/scala/spark/executor/TaskMetrics.scala @@ -49,11 +49,6 @@ class ShuffleReadMetrics extends Serializable { var localBlocksFetched: Int = _ /** - * Total time to read shuffle data - */ - var shuffleReadMillis: Long = _ - - /** * Total time that is spent blocked waiting for shuffle to fetch data */ var fetchWaitTime: Long = _ diff --git a/core/src/main/scala/spark/storage/DelegateBlockFetchTracker.scala b/core/src/main/scala/spark/storage/DelegateBlockFetchTracker.scala deleted file mode 100644 index f6c28dce52..0000000000 --- a/core/src/main/scala/spark/storage/DelegateBlockFetchTracker.scala +++ /dev/null @@ -1,12 +0,0 @@ -package spark.storage - -private[spark] trait DelegateBlockFetchTracker extends BlockFetchTracker { - var delegate : BlockFetchTracker = _ - def setDelegate(d: BlockFetchTracker) {delegate = d} - def totalBlocks = delegate.totalBlocks - def numLocalBlocks = delegate.numLocalBlocks - def numRemoteBlocks = delegate.numRemoteBlocks - def remoteFetchTime = delegate.remoteFetchTime - def fetchWaitTime = delegate.fetchWaitTime - def remoteBytesRead = delegate.remoteBytesRead -} diff --git a/core/src/main/scala/spark/util/TimedIterator.scala b/core/src/main/scala/spark/util/TimedIterator.scala deleted file mode 100644 index 539b01f4ce..0000000000 --- a/core/src/main/scala/spark/util/TimedIterator.scala +++ /dev/null @@ -1,32 +0,0 @@ -package spark.util - -/** - * A utility for tracking the total time an iterator takes to iterate through its elements. - * - * In general, this should only be used if you expect it to take a considerable amount of time - * (eg. milliseconds) to get each element -- otherwise, the timing won't be very accurate, - * and you are probably just adding more overhead - */ -class TimedIterator[+A](val sub: Iterator[A]) extends Iterator[A] { - private var netMillis = 0l - private var nElems = 0 - def hasNext = { - val start = System.currentTimeMillis() - val r = sub.hasNext - val end = System.currentTimeMillis() - netMillis += (end - start) - r - } - def next = { - val start = System.currentTimeMillis() - val r = sub.next - val end = System.currentTimeMillis() - netMillis += (end - start) - nElems += 1 - r - } - - def getNetMillis = netMillis - def getAverageTimePerItem = netMillis / nElems.toDouble - -} |