diff --git a/core/src/main/scala/org/apache/spark/Accumulators.scala b/core/src/main/scala/org/apache/spark/Accumulators.scala
index 9c55bfbb47626..12f2fe031cb1d 100644
--- a/core/src/main/scala/org/apache/spark/Accumulators.scala
+++ b/core/src/main/scala/org/apache/spark/Accumulators.scala
@@ -36,15 +36,21 @@ import org.apache.spark.serializer.JavaSerializer
*
* @param initialValue initial value of accumulator
* @param param helper object defining how to add elements of type `R` and `T`
+ * @param name human-readable name for use in Spark's web UI
* @tparam R the full accumulated data (result type)
* @tparam T partial data that can be added in
*/
class Accumulable[R, T] (
@transient initialValue: R,
- param: AccumulableParam[R, T])
+ param: AccumulableParam[R, T],
+ val name: Option[String])
extends Serializable {
- val id = Accumulators.newId
+ def this(@transient initialValue: R, param: AccumulableParam[R, T]) =
+ this(initialValue, param, None)
+
+ val id: Long = Accumulators.newId
+
@transient private var value_ = initialValue // Current value on master
val zero = param.zero(initialValue) // Zero value to be passed to workers
private var deserialized = false
@@ -219,8 +225,10 @@ GrowableAccumulableParam[R <% Growable[T] with TraversableOnce[T] with Serializa
* @param param helper object defining how to add elements of type `T`
* @tparam T result type
*/
-class Accumulator[T](@transient initialValue: T, param: AccumulatorParam[T])
- extends Accumulable[T,T](initialValue, param)
+class Accumulator[T](@transient initialValue: T, param: AccumulatorParam[T], name: Option[String])
+ extends Accumulable[T,T](initialValue, param, name) {
+ def this(initialValue: T, param: AccumulatorParam[T]) = this(initialValue, param, None)
+}
/**
* A simpler version of [[org.apache.spark.AccumulableParam]] where the only data type you can add
@@ -281,4 +289,7 @@ private object Accumulators {
}
}
}
+
+ def stringifyPartialValue(partialValue: Any) = "%s".format(partialValue)
+ def stringifyValue(value: Any) = "%s".format(value)
}
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 9ba21cfcde01a..e132955f0f850 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -760,6 +760,15 @@ class SparkContext(config: SparkConf) extends Logging {
def accumulator[T](initialValue: T)(implicit param: AccumulatorParam[T]) =
new Accumulator(initialValue, param)
+ /**
+ * Create an [[org.apache.spark.Accumulator]] variable of a given type, with a name for display
+ * in the Spark UI. Tasks can "add" values to the accumulator using the `+=` method. Only the
+ * driver can access the accumulator's `value`.
+ */
+ def accumulator[T](initialValue: T, name: String)(implicit param: AccumulatorParam[T]) = {
+ new Accumulator(initialValue, param, Some(name))
+ }
+
/**
* Create an [[org.apache.spark.Accumulable]] shared variable, to which tasks can add values
* with `+=`. Only the driver can access the accumuable's `value`.
@@ -769,6 +778,16 @@ class SparkContext(config: SparkConf) extends Logging {
def accumulable[T, R](initialValue: T)(implicit param: AccumulableParam[T, R]) =
new Accumulable(initialValue, param)
+ /**
+ * Create an [[org.apache.spark.Accumulable]] shared variable, with a name for display in the
+ * Spark UI. Tasks can add values to the accumuable using the `+=` operator. Only the driver can
+ * access the accumuable's `value`.
+ * @tparam T accumulator type
+ * @tparam R type that can be added to the accumulator
+ */
+ def accumulable[T, R](initialValue: T, name: String)(implicit param: AccumulableParam[T, R]) =
+ new Accumulable(initialValue, param, Some(name))
+
/**
* Create an accumulator from a "mutable collection" type.
*
diff --git a/core/src/main/scala/org/apache/spark/api/java/JavaSparkContext.scala b/core/src/main/scala/org/apache/spark/api/java/JavaSparkContext.scala
index d9d1c5955ca99..e0a4815940db3 100644
--- a/core/src/main/scala/org/apache/spark/api/java/JavaSparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/api/java/JavaSparkContext.scala
@@ -429,6 +429,16 @@ class JavaSparkContext(val sc: SparkContext) extends JavaSparkContextVarargsWork
def intAccumulator(initialValue: Int): Accumulator[java.lang.Integer] =
sc.accumulator(initialValue)(IntAccumulatorParam).asInstanceOf[Accumulator[java.lang.Integer]]
+ /**
+ * Create an [[org.apache.spark.Accumulator]] integer variable, which tasks can "add" values
+ * to using the `add` method. Only the master can access the accumulator's `value`.
+ *
+ * This version supports naming the accumulator for display in Spark's web UI.
+ */
+ def intAccumulator(initialValue: Int, name: String): Accumulator[java.lang.Integer] =
+ sc.accumulator(initialValue, name)(IntAccumulatorParam)
+ .asInstanceOf[Accumulator[java.lang.Integer]]
+
/**
* Create an [[org.apache.spark.Accumulator]] double variable, which tasks can "add" values
* to using the `add` method. Only the master can access the accumulator's `value`.
@@ -436,12 +446,31 @@ class JavaSparkContext(val sc: SparkContext) extends JavaSparkContextVarargsWork
def doubleAccumulator(initialValue: Double): Accumulator[java.lang.Double] =
sc.accumulator(initialValue)(DoubleAccumulatorParam).asInstanceOf[Accumulator[java.lang.Double]]
+ /**
+ * Create an [[org.apache.spark.Accumulator]] double variable, which tasks can "add" values
+ * to using the `add` method. Only the master can access the accumulator's `value`.
+ *
+ * This version supports naming the accumulator for display in Spark's web UI.
+ */
+ def doubleAccumulator(initialValue: Double, name: String): Accumulator[java.lang.Double] =
+ sc.accumulator(initialValue, name)(DoubleAccumulatorParam)
+ .asInstanceOf[Accumulator[java.lang.Double]]
+
/**
* Create an [[org.apache.spark.Accumulator]] integer variable, which tasks can "add" values
* to using the `add` method. Only the master can access the accumulator's `value`.
*/
def accumulator(initialValue: Int): Accumulator[java.lang.Integer] = intAccumulator(initialValue)
+ /**
+ * Create an [[org.apache.spark.Accumulator]] integer variable, which tasks can "add" values
+ * to using the `add` method. Only the master can access the accumulator's `value`.
+ *
+ * This version supports naming the accumulator for display in Spark's web UI.
+ */
+ def accumulator(initialValue: Int, name: String): Accumulator[java.lang.Integer] =
+ intAccumulator(initialValue, name)
+
/**
* Create an [[org.apache.spark.Accumulator]] double variable, which tasks can "add" values
* to using the `add` method. Only the master can access the accumulator's `value`.
@@ -449,6 +478,16 @@ class JavaSparkContext(val sc: SparkContext) extends JavaSparkContextVarargsWork
def accumulator(initialValue: Double): Accumulator[java.lang.Double] =
doubleAccumulator(initialValue)
+
+ /**
+ * Create an [[org.apache.spark.Accumulator]] double variable, which tasks can "add" values
+ * to using the `add` method. Only the master can access the accumulator's `value`.
+ *
+ * This version supports naming the accumulator for display in Spark's web UI.
+ */
+ def accumulator(initialValue: Double, name: String): Accumulator[java.lang.Double] =
+ doubleAccumulator(initialValue, name)
+
/**
* Create an [[org.apache.spark.Accumulator]] variable of a given type, which tasks can "add"
* values to using the `add` method. Only the master can access the accumulator's `value`.
@@ -456,6 +495,16 @@ class JavaSparkContext(val sc: SparkContext) extends JavaSparkContextVarargsWork
def accumulator[T](initialValue: T, accumulatorParam: AccumulatorParam[T]): Accumulator[T] =
sc.accumulator(initialValue)(accumulatorParam)
+ /**
+ * Create an [[org.apache.spark.Accumulator]] variable of a given type, which tasks can "add"
+ * values to using the `add` method. Only the master can access the accumulator's `value`.
+ *
+ * This version supports naming the accumulator for display in Spark's web UI.
+ */
+ def accumulator[T](initialValue: T, name: String, accumulatorParam: AccumulatorParam[T])
+ : Accumulator[T] =
+ sc.accumulator(initialValue, name)(accumulatorParam)
+
/**
* Create an [[org.apache.spark.Accumulable]] shared variable of the given type, to which tasks
* can "add" values with `add`. Only the master can access the accumuable's `value`.
@@ -463,6 +512,16 @@ class JavaSparkContext(val sc: SparkContext) extends JavaSparkContextVarargsWork
def accumulable[T, R](initialValue: T, param: AccumulableParam[T, R]): Accumulable[T, R] =
sc.accumulable(initialValue)(param)
+ /**
+ * Create an [[org.apache.spark.Accumulable]] shared variable of the given type, to which tasks
+ * can "add" values with `add`. Only the master can access the accumuable's `value`.
+ *
+ * This version supports naming the accumulator for display in Spark's web UI.
+ */
+ def accumulable[T, R](initialValue: T, name: String, param: AccumulableParam[T, R])
+ : Accumulable[T, R] =
+ sc.accumulable(initialValue, name)(param)
+
/**
* Broadcast a read-only variable to the cluster, returning a
* [[org.apache.spark.broadcast.Broadcast]] object for reading it in distributed functions.
diff --git a/core/src/main/scala/org/apache/spark/scheduler/AccumulableInfo.scala b/core/src/main/scala/org/apache/spark/scheduler/AccumulableInfo.scala
new file mode 100644
index 0000000000000..fa83372bb4d11
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/scheduler/AccumulableInfo.scala
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler
+
+import org.apache.spark.annotation.DeveloperApi
+
+/**
+ * :: DeveloperApi ::
+ * Information about an [[org.apache.spark.Accumulable]] modified during a task or stage.
+ */
+@DeveloperApi
+class AccumulableInfo (
+ val id: Long,
+ val name: String,
+ val update: Option[String], // represents a partial update within a task
+ val value: String) {
+
+ override def equals(other: Any): Boolean = other match {
+ case acc: AccumulableInfo =>
+ this.id == acc.id && this.name == acc.name &&
+ this.update == acc.update && this.value == acc.value
+ case _ => false
+ }
+}
+
+object AccumulableInfo {
+ def apply(id: Long, name: String, update: Option[String], value: String) =
+ new AccumulableInfo(id, name, update, value)
+
+ def apply(id: Long, name: String, value: String) = new AccumulableInfo(id, name, None, value)
+}
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 9fa3a4e9c71ae..430e45ada5808 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -883,8 +883,14 @@ class DAGScheduler(
val task = event.task
val stageId = task.stageId
val taskType = Utils.getFormattedClassName(task)
- listenerBus.post(SparkListenerTaskEnd(stageId, taskType, event.reason, event.taskInfo,
- event.taskMetrics))
+
+ // The success case is dealt with separately below, since we need to compute accumulator
+ // updates before posting.
+ if (event.reason != Success) {
+ listenerBus.post(SparkListenerTaskEnd(stageId, taskType, event.reason, event.taskInfo,
+ event.taskMetrics))
+ }
+
if (!stageIdToStage.contains(task.stageId)) {
// Skip all the actions if the stage has been cancelled.
return
@@ -906,12 +912,26 @@ class DAGScheduler(
if (event.accumUpdates != null) {
try {
Accumulators.add(event.accumUpdates)
+ event.accumUpdates.foreach { case (id, partialValue) =>
+ val acc = Accumulators.originals(id).asInstanceOf[Accumulable[Any, Any]]
+ // To avoid UI cruft, ignore cases where value wasn't updated
+ if (acc.name.isDefined && partialValue != acc.zero) {
+ val name = acc.name.get
+ val stringPartialValue = Accumulators.stringifyPartialValue(partialValue)
+ val stringValue = Accumulators.stringifyValue(acc.value)
+ stage.info.accumulables(id) = AccumulableInfo(id, name, stringValue)
+ event.taskInfo.accumulables +=
+ AccumulableInfo(id, name, Some(stringPartialValue), stringValue)
+ }
+ }
} catch {
// If we see an exception during accumulator update, just log the error and move on.
case e: Exception =>
logError(s"Failed to update accumulators for $task", e)
}
}
+ listenerBus.post(SparkListenerTaskEnd(stageId, taskType, event.reason, event.taskInfo,
+ event.taskMetrics))
stage.pendingTasks -= task
task match {
case rt: ResultTask[_, _] =>
diff --git a/core/src/main/scala/org/apache/spark/scheduler/StageInfo.scala b/core/src/main/scala/org/apache/spark/scheduler/StageInfo.scala
index 480891550eb60..2a407e47a05bd 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/StageInfo.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/StageInfo.scala
@@ -17,6 +17,8 @@
package org.apache.spark.scheduler
+import scala.collection.mutable.HashMap
+
import org.apache.spark.annotation.DeveloperApi
import org.apache.spark.storage.RDDInfo
@@ -37,6 +39,8 @@ class StageInfo(
var completionTime: Option[Long] = None
/** If the stage failed, the reason why. */
var failureReason: Option[String] = None
+ /** Terminal values of accumulables updated during this stage. */
+ val accumulables = HashMap[Long, AccumulableInfo]()
def stageFailed(reason: String) {
failureReason = Some(reason)
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskInfo.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskInfo.scala
index ca0595f35143e..6fa1f2c880f7a 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskInfo.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskInfo.scala
@@ -17,6 +17,8 @@
package org.apache.spark.scheduler
+import scala.collection.mutable.ListBuffer
+
import org.apache.spark.annotation.DeveloperApi
/**
@@ -41,6 +43,13 @@ class TaskInfo(
*/
var gettingResultTime: Long = 0
+ /**
+ * Intermediate updates to accumulables during this task. Note that it is valid for the same
+ * accumulable to be updated multiple times in a single task or for two accumulables with the
+ * same name but different IDs to exist in a task.
+ */
+ val accumulables = ListBuffer[AccumulableInfo]()
+
/**
* The time when the task has completed successfully (including the time to remotely fetch
* results, if necessary).
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 da2f5d3172fe2..a57a354620163 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
@@ -17,7 +17,7 @@
package org.apache.spark.ui.jobs
-import scala.collection.mutable.{HashMap, ListBuffer}
+import scala.collection.mutable.{HashMap, ListBuffer, Map}
import org.apache.spark._
import org.apache.spark.annotation.DeveloperApi
@@ -65,6 +65,10 @@ class JobProgressListener(conf: SparkConf) extends SparkListener with Logging {
new StageUIData
})
+ for ((id, info) <- stageCompleted.stageInfo.accumulables) {
+ stageData.accumulables(id) = info
+ }
+
poolToActiveStages.get(stageData.schedulingPool).foreach(_.remove(stageId))
activeStages.remove(stageId)
if (stage.failureReason.isEmpty) {
@@ -130,6 +134,10 @@ class JobProgressListener(conf: SparkConf) extends SparkListener with Logging {
new StageUIData
})
+ for (accumulableInfo <- info.accumulables) {
+ stageData.accumulables(accumulableInfo.id) = accumulableInfo
+ }
+
val execSummaryMap = stageData.executorSummary
val execSummary = execSummaryMap.getOrElseUpdate(info.executorId, new ExecutorSummary)
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala
index cab26b9e2f7d3..8bc1ba758cf77 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala
@@ -20,11 +20,12 @@ package org.apache.spark.ui.jobs
import java.util.Date
import javax.servlet.http.HttpServletRequest
-import scala.xml.Node
+import scala.xml.{Node, Unparsed}
import org.apache.spark.ui.{ToolTips, WebUIPage, UIUtils}
import org.apache.spark.ui.jobs.UIData._
import org.apache.spark.util.{Utils, Distribution}
+import org.apache.spark.scheduler.AccumulableInfo
/** Page showing statistics and task list for a given stage */
private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
@@ -51,6 +52,7 @@ private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
val tasks = stageData.taskData.values.toSeq.sortBy(_.taskInfo.launchTime)
val numCompleted = tasks.count(_.taskInfo.finished)
+ val accumulables = listener.stageIdToData(stageId).accumulables
val hasInput = stageData.inputBytes > 0
val hasShuffleRead = stageData.shuffleReadBytes > 0
val hasShuffleWrite = stageData.shuffleWriteBytes > 0
@@ -95,10 +97,15 @@ private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
// scalastyle:on
+ val accumulableHeaders: Seq[String] = Seq("Accumulable", "Value")
+ def accumulableRow(acc: AccumulableInfo) = | {acc.name} | {acc.value} |
+ val accumulableTable = UIUtils.listingTable(accumulableHeaders, accumulableRow,
+ accumulables.values.toSeq)
+
val taskHeaders: Seq[String] =
Seq(
"Index", "ID", "Attempt", "Status", "Locality Level", "Executor",
- "Launch Time", "Duration", "GC Time") ++
+ "Launch Time", "Duration", "GC Time", "Accumulators") ++
{if (hasInput) Seq("Input") else Nil} ++
{if (hasShuffleRead) Seq("Shuffle Read") else Nil} ++
{if (hasShuffleWrite) Seq("Write Time", "Shuffle Write") else Nil} ++
@@ -208,11 +215,16 @@ private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
Some(UIUtils.listingTable(quantileHeaders, quantileRow, listings, fixedWidth = true))
}
val executorTable = new ExecutorTable(stageId, parent)
+
+ val maybeAccumulableTable: Seq[Node] =
+ if (accumulables.size > 0) { Accumulators
++ accumulableTable } else Seq()
+
val content =
summary ++
Summary Metrics for {numCompleted} Completed Tasks
++
{summaryTable.getOrElse("No tasks have reported metrics yet.")}
++
Aggregated Metrics by Executor
++ executorTable.toNodeSeq ++
+ maybeAccumulableTable ++
Tasks
++ taskTable
UIUtils.headerSparkPage(content, basePath, appName, "Details for Stage %d".format(stageId),
@@ -279,6 +291,11 @@ private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
{if (gcTime > 0) UIUtils.formatDuration(gcTime) else ""}
|
+
+ {Unparsed(
+ info.accumulables.map{acc => s"${acc.name}: ${acc.update.get}"}.mkString("
")
+ )}
+ |