From 37d925280cdfdda8f6f7174c67a614056eea5d69 Mon Sep 17 00:00:00 2001
From: yantangzhai <tyz0303@163.com>
Date: Mon, 15 Sep 2014 16:57:38 -0700
Subject: [PATCH] [SPARK-2714] DAGScheduler logs jobid when runJob finishes
DAGScheduler logs jobid when runJob finishes
Author: yantangzhai <tyz0303@163.com>
Closes #1617 from YanTangZhai/SPARK-2714 and squashes the following commits:
0a0243f [yantangzhai] [SPARK-2714] DAGScheduler logs jobid when runJob finishes
fbb1150 [yantangzhai] [SPARK-2714] DAGScheduler logs jobid when runJob finishes
7aec2a9 [yantangzhai] [SPARK-2714] DAGScheduler logs jobid when runJob finishes
fb42f0f [yantangzhai] [SPARK-2714] DAGScheduler logs jobid when runJob finishes
090d908 [yantangzhai] [SPARK-2714] DAGScheduler logs jobid when runJob finishes
---
core/src/main/scala/org/apache/spark/SparkContext.scala | 3 ---
.../scala/org/apache/spark/scheduler/DAGScheduler.scala | 9 +++++++--
2 files changed, 7 insertions(+), 5 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 218b353dd9..428f019b02 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -1072,11 +1072,8 @@ class SparkContext(config: SparkConf) extends Logging {
val callSite = getCallSite
val cleanedFunc = clean(func)
logInfo("Starting job: " + callSite.shortForm)
- val start = System.nanoTime
dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, allowLocal,
resultHandler, localProperties.get)
- logInfo(
- "Job finished: " + callSite.shortForm + ", took " + (System.nanoTime - start) / 1e9 + " s")
rdd.doCheckpoint()
}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 6fcf9e3154..b2774dfc47 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -507,11 +507,16 @@ class DAGScheduler(
resultHandler: (Int, U) => Unit,
properties: Properties = null)
{
+ val start = System.nanoTime
val waiter = submitJob(rdd, func, partitions, callSite, allowLocal, resultHandler, properties)
waiter.awaitResult() match {
- case JobSucceeded => {}
+ case JobSucceeded => {
+ logInfo("Job %d finished: %s, took %f s".format
+ (waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
+ }
case JobFailed(exception: Exception) =>
- logInfo("Failed to run " + callSite.shortForm)
+ logInfo("Job %d failed: %s, took %f s".format
+ (waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
throw exception
}
}
--
GitLab