apache/incubator-spark — pull request #422 by sproblvem (2014-01-14 17:25:42+00:00)
Title: modify metrics to profile spark more accurate

This pull request include 3 modifications, try to profile spark more accurate. I think it's also helpfull for others.

0. Unify the count unit of "shuffleWriteTime" to millisecond
The count unit of "shuffleWriteTime" was nanosecond while other metrics was count in millisecond.
The original code will print 6 time related terms. In JobLogger.scala their variable name are
"taskInfo.launchTime, taskInfo.finishTime, metrics.shuffleFinishTime" and
"taskMetrics.executorRunTime, metrics.fetchWaitTime, metrics.remoteFetchTime".
The first three vairables were assigned by the return value of function call "System.currentTimeMillis()" directly, the second three variables were calculate also in millisecond. Here are the code of calculating "taskMetrics.executorRunTime"

taskStart = System.currentTimeMillis()
val value = task.run(taskId.toInt)
val taskFinish = System.currentTimeMillis()
m.executorRunTime = (taskFinish - taskStart).toInt

But the code of calculating shuffleWriteTime were:

private def callWithTiming(f: => Unit) = {
val start = System.nanoTime()
f
_timeWriting += (System.nanoTime() - start)
}

def timeWriting = _timeWriting

totalTime += writer.timeWriting()

shuffleMetrics.shuffleWriteTime = totalTime

So I just divide "totalTime" by 1000000 to cast it into millisecondes. Then assign it to shuffleWriteTime.

1. Include the running time of function "flush" to "shuffleWriteTime"
Write key-value pairs to disk include two parts, first write every kv pairs into buffer, then flush the buffer to disk. The original code only include the total time of write kv paris to buffer, so I rewrite the flush function to include its running time.

2. Report the running time of tasks on worker
When I profile spark, I found that the "taskInfo.launchTime, taskInfo.finishTime" are not enough to observe the exact behavior of spark. For cluster which doesn't set NTP, the clock time of different worker will differ in seconds. But these "taskInfo.launchTime, taskInfo.finishTime" are both colleceted on driver, we can't use these two metrics to get the exactly start / finish time of tasks on worker.
So I add two member variables for class "TaskMetrics" to record the start time and finish time of each task on worker, then modify JobLogger to report them.