You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by va...@apache.org on 2017/03/31 16:43:02 UTC
spark git commit: [SPARK-20084][CORE] Remove
internal.metrics.updatedBlockStatuses from history files.
Repository: spark
Updated Branches:
refs/heads/master 254877c2f -> c4c03eed6
[SPARK-20084][CORE] Remove internal.metrics.updatedBlockStatuses from history files.
## What changes were proposed in this pull request?
Remove accumulator updates for internal.metrics.updatedBlockStatuses from SparkListenerTaskEnd entries in the history file. These can cause history files to grow to hundreds of GB because the value of the accumulator contains all tracked blocks.
## How was this patch tested?
Current History UI tests cover use of the history file.
Author: Ryan Blue <bl...@apache.org>
Closes #17412 from rdblue/SPARK-20084-remove-block-accumulator-info.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/c4c03eed
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/c4c03eed
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/c4c03eed
Branch: refs/heads/master
Commit: c4c03eed67c05a78dc8944f6119ea708d6b955be
Parents: 254877c
Author: Ryan Blue <bl...@apache.org>
Authored: Fri Mar 31 09:42:49 2017 -0700
Committer: Marcelo Vanzin <va...@cloudera.com>
Committed: Fri Mar 31 09:42:49 2017 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/util/JsonProtocol.scala | 15 +++++++++++----
1 file changed, 11 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/c4c03eed/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala b/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala
index 2cb8891..1d2cb7a 100644
--- a/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala
+++ b/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala
@@ -264,8 +264,7 @@ private[spark] object JsonProtocol {
("Submission Time" -> submissionTime) ~
("Completion Time" -> completionTime) ~
("Failure Reason" -> failureReason) ~
- ("Accumulables" -> JArray(
- stageInfo.accumulables.values.map(accumulableInfoToJson).toList))
+ ("Accumulables" -> accumulablesToJson(stageInfo.accumulables.values))
}
def taskInfoToJson(taskInfo: TaskInfo): JValue = {
@@ -281,7 +280,15 @@ private[spark] object JsonProtocol {
("Finish Time" -> taskInfo.finishTime) ~
("Failed" -> taskInfo.failed) ~
("Killed" -> taskInfo.killed) ~
- ("Accumulables" -> JArray(taskInfo.accumulables.toList.map(accumulableInfoToJson)))
+ ("Accumulables" -> accumulablesToJson(taskInfo.accumulables))
+ }
+
+ private lazy val accumulableBlacklist = Set("internal.metrics.updatedBlockStatuses")
+
+ def accumulablesToJson(accumulables: Traversable[AccumulableInfo]): JArray = {
+ JArray(accumulables
+ .filterNot(_.name.exists(accumulableBlacklist.contains))
+ .toList.map(accumulableInfoToJson))
}
def accumulableInfoToJson(accumulableInfo: AccumulableInfo): JValue = {
@@ -376,7 +383,7 @@ private[spark] object JsonProtocol {
("Message" -> fetchFailed.message)
case exceptionFailure: ExceptionFailure =>
val stackTrace = stackTraceToJson(exceptionFailure.stackTrace)
- val accumUpdates = JArray(exceptionFailure.accumUpdates.map(accumulableInfoToJson).toList)
+ val accumUpdates = accumulablesToJson(exceptionFailure.accumUpdates)
("Class Name" -> exceptionFailure.className) ~
("Description" -> exceptionFailure.description) ~
("Stack Trace" -> stackTrace) ~
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org