diff options
author | Wenchen Fan <wenchen@databricks.com> | 2016-05-10 11:16:31 -0700 |
---|---|---|
committer | Andrew Or <andrew@databricks.com> | 2016-05-10 11:16:56 -0700 |
commit | bcfee153b1cacfe617e602f3b72c0877e0bdf1f7 (patch) | |
tree | 7d8aa24d2213d098901547d3b81390fb67bef2be /sql/core | |
parent | 0b9cae42426e14060bc6182c037fd715f35a2d23 (diff) | |
download | spark-bcfee153b1cacfe617e602f3b72c0877e0bdf1f7.tar.gz spark-bcfee153b1cacfe617e602f3b72c0877e0bdf1f7.tar.bz2 spark-bcfee153b1cacfe617e602f3b72c0877e0bdf1f7.zip |
[SPARK-12837][CORE] reduce network IO for accumulators
Sending un-updated accumulators back to driver makes no sense, as merging a zero value accumulator is a no-op. We should only send back updated accumulators, to save network IO.
new test in `TaskContextSuite`
Author: Wenchen Fan <wenchen@databricks.com>
Closes #12899 from cloud-fan/acc.
Diffstat (limited to 'sql/core')
3 files changed, 10 insertions, 10 deletions
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLMetrics.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLMetrics.scala index f82e0b8bca..786110477d 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLMetrics.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLMetrics.scala @@ -66,7 +66,7 @@ private[sql] object SQLMetrics { def createMetric(sc: SparkContext, name: String): SQLMetric = { val acc = new SQLMetric(SUM_METRIC) - acc.register(sc, name = Some(name), countFailedValues = true) + acc.register(sc, name = Some(name), countFailedValues = false) acc } @@ -79,7 +79,7 @@ private[sql] object SQLMetrics { // data size total (min, med, max): // 100GB (100MB, 1GB, 10GB) val acc = new SQLMetric(SIZE_METRIC, -1) - acc.register(sc, name = Some(s"$name total (min, med, max)"), countFailedValues = true) + acc.register(sc, name = Some(s"$name total (min, med, max)"), countFailedValues = false) acc } @@ -88,7 +88,7 @@ private[sql] object SQLMetrics { // duration(min, med, max): // 5s (800ms, 1s, 2s) val acc = new SQLMetric(TIMING_METRIC, -1) - acc.register(sc, name = Some(s"$name total (min, med, max)"), countFailedValues = true) + acc.register(sc, name = Some(s"$name total (min, med, max)"), countFailedValues = false) acc } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala index 29c54111ea..510a2ee3bf 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala @@ -164,7 +164,7 @@ private[sql] class SQLListener(conf: SparkConf) extends SparkListener with Loggi taskEnd.taskInfo.taskId, taskEnd.stageId, taskEnd.stageAttemptId, - taskEnd.taskMetrics.accumulators().map(a => a.toInfo(Some(a.value), None)), + taskEnd.taskMetrics.externalAccums.map(a => a.toInfo(Some(a.value), None)), finishTask = true) } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLListenerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLListenerSuite.scala index 5e08658e5e..67e44849ca 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLListenerSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLListenerSuite.scala @@ -19,7 +19,7 @@ package org.apache.spark.sql.execution.ui import java.util.Properties -import org.mockito.Mockito.{mock, when} +import org.mockito.Mockito.mock import org.apache.spark._ import org.apache.spark.executor.TaskMetrics @@ -74,13 +74,13 @@ class SQLListenerSuite extends SparkFunSuite with SharedSQLContext { ) private def createTaskMetrics(accumulatorUpdates: Map[Long, Long]): TaskMetrics = { - val metrics = mock(classOf[TaskMetrics]) - when(metrics.accumulators()).thenReturn(accumulatorUpdates.map { case (id, update) => + val metrics = TaskMetrics.empty + accumulatorUpdates.foreach { case (id, update) => val acc = new LongAccumulator acc.metadata = AccumulatorMetadata(id, Some(""), true) - acc.setValue(update) - acc - }.toSeq) + acc.add(update) + metrics.registerAccumulator(acc) + } metrics } |