From 7875a49a234f8c9055991faccd5a5a9a7f470fa4 Mon Sep 17 00:00:00 2001
From: "navis.ryu" <navis@apache.org>
Date: Tue, 5 Jan 2016 10:49:27 +0900
Subject: [PATCH] SPARK-12637 Print stage info of finished stages properly

---
 .../org/apache/spark/scheduler/SparkListener.scala    | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala b/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala
index 3130a65240a9..094ee821f1c8 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala
@@ -266,7 +266,7 @@ class StatsReportListener extends SparkListener with Logging {
 
   override def onStageCompleted(stageCompleted: SparkListenerStageCompleted) {
     implicit val sc = stageCompleted
-    this.logInfo("Finished stage: " + stageCompleted.stageInfo)
+    this.logInfo("Finished stage: " + getStatusDetail(stageCompleted.stageInfo))
     showMillisDistribution("task runtime:", (info, _) => Some(info.duration), taskInfoMetrics)
 
     // Shuffle write
@@ -293,6 +293,15 @@ class StatsReportListener extends SparkListener with Logging {
     taskInfoMetrics.clear()
   }
 
+  private[spark] def getStatusDetail(stageInfo: StageInfo): String = {
+    "Stage(%d, %d); Name: \"%s\"; Status: %s%s; numTasks: %d; Took: %s msec".format(
+      stageInfo.stageId, stageInfo.attemptId, stageInfo.name, stageInfo.getStatusString,
+      stageInfo.failureReason.map(x => "(" + x + ")").getOrElse(""),
+      stageInfo.numTasks,
+      stageInfo.submissionTime.map(
+        x => stageInfo.completionTime.getOrElse(System.currentTimeMillis()) - x).getOrElse("-")
+    )
+  }
 }
 
 private[spark] object StatsReportListener extends Logging {