You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by rx...@apache.org on 2013/12/23 19:38:48 UTC
[6/9] git commit: spark-968, changes for avoid a NPE
spark-968, changes for avoid a NPE
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/59e53fa2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/59e53fa2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/59e53fa2
Branch: refs/heads/master
Commit: 59e53fa21caa202a57093c74ada128fca2be5bac
Parents: 36060f4
Author: wangda.tan <wh...@gmail.com>
Authored: Tue Dec 17 17:57:27 2013 +0800
Committer: wangda.tan <wh...@gmail.com>
Committed: Tue Dec 17 17:57:27 2013 +0800
----------------------------------------------------------------------
.../org/apache/spark/ui/exec/ExecutorsUI.scala | 30 +++++++++++---------
.../spark/ui/jobs/JobProgressListener.scala | 24 +++++++++-------
2 files changed, 29 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/59e53fa2/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsUI.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsUI.scala b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsUI.scala
index 808bbe8..f62ae37 100644
--- a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsUI.scala
+++ b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsUI.scala
@@ -150,7 +150,7 @@ private[spark] class ExecutorsUI(val sc: SparkContext) {
activeTasks += taskStart.taskInfo
}
- override def onTaskEnd(taskEnd: SparkListenerTaskEnd) = synchronized {
+ override def onTaskEnd(taskEnd: SparkListenerTaskEnd) {
val eid = taskEnd.taskInfo.executorId
val activeTasks = executorToTasksActive.getOrElseUpdate(eid, new HashSet[TaskInfo]())
val newDuration = executorToDuration.getOrElse(eid, 0L) + taskEnd.taskInfo.duration
@@ -168,20 +168,22 @@ private[spark] class ExecutorsUI(val sc: SparkContext) {
}
// update shuffle read/write
- val shuffleRead = taskEnd.taskMetrics.shuffleReadMetrics
- shuffleRead match {
- case Some(s) =>
- val newShuffleRead = executorToShuffleRead.getOrElse(eid, 0L) + s.remoteBytesRead
- executorToShuffleRead.put(eid, newShuffleRead)
- case _ => {}
- }
- val shuffleWrite = taskEnd.taskMetrics.shuffleWriteMetrics
- shuffleWrite match {
- case Some(s) => {
- val newShuffleWrite = executorToShuffleWrite.getOrElse(eid, 0L) + s.shuffleBytesWritten
- executorToShuffleWrite.put(eid, newShuffleWrite)
+ if (null != taskEnd.taskMetrics) {
+ val shuffleRead = taskEnd.taskMetrics.shuffleReadMetrics
+ shuffleRead match {
+ case Some(s) =>
+ val newShuffleRead = executorToShuffleRead.getOrElse(eid, 0L) + s.remoteBytesRead
+ executorToShuffleRead.put(eid, newShuffleRead)
+ case _ => {}
+ }
+ val shuffleWrite = taskEnd.taskMetrics.shuffleWriteMetrics
+ shuffleWrite match {
+ case Some(s) => {
+ val newShuffleWrite = executorToShuffleWrite.getOrElse(eid, 0L) + s.shuffleBytesWritten
+ executorToShuffleWrite.put(eid, newShuffleWrite)
+ }
+ case _ => {}
}
- case _ => {}
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/59e53fa2/core/src/main/scala/org/apache/spark/ui/jobs/JobProgressListener.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/JobProgressListener.scala b/core/src/main/scala/org/apache/spark/ui/jobs/JobProgressListener.scala
index 8c92ff1..64ce715 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/JobProgressListener.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/JobProgressListener.scala
@@ -147,18 +147,20 @@ private[spark] class JobProgressListener(val sc: SparkContext) extends SparkList
y.duration += taskEnd.taskInfo.duration
// update shuffle read/write
- val shuffleRead = taskEnd.taskMetrics.shuffleReadMetrics
- shuffleRead match {
- case Some(s) =>
- y.shuffleRead += s.remoteBytesRead
- case _ => {}
- }
- val shuffleWrite = taskEnd.taskMetrics.shuffleWriteMetrics
- shuffleWrite match {
- case Some(s) => {
- y.shuffleWrite += s.shuffleBytesWritten
+ if (null != taskEnd.taskMetrics) {
+ val shuffleRead = taskEnd.taskMetrics.shuffleReadMetrics
+ shuffleRead match {
+ case Some(s) =>
+ y.shuffleRead += s.remoteBytesRead
+ case _ => {}
+ }
+ val shuffleWrite = taskEnd.taskMetrics.shuffleWriteMetrics
+ shuffleWrite match {
+ case Some(s) => {
+ y.shuffleWrite += s.shuffleBytesWritten
+ }
+ case _ => {}
}
- case _ => {}
}
}
case _ => {}