You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by advancedxy <gi...@git.apache.org> on 2018/04/26 09:28:34 UTC
[GitHub] spark pull request #21165: Spark 20087: Attach accumulators / metrics to 'Ta...
GitHub user advancedxy opened a pull request:
https://github.com/apache/spark/pull/21165
Spark 20087: Attach accumulators / metrics to 'TaskKilled' end reason
## What changes were proposed in this pull request?
The ultimate goal is for listeners to onTaskEnd to receive metrics when a task is killed intentionally, since the data is currently just thrown away. This is already done for ExceptionFailure, so this just copies the same approach.
## How was this patch tested?
Updated existing tests.
This is a rework of https://github.com/apache/spark/pull/17422, all credits should go to @noodle-fb
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/advancedxy/spark SPARK-20087
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/21165.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #21165
----
commit 6262c52f7eb4abcf06742e9afd9d1454f06cdf1f
Author: Charles Lewis <no...@...>
Date: 2017-03-22T20:33:55Z
report metrics for killed tasks
commit 18308895ad07c9c757e96e08f5c40b5dcaaf3455
Author: Charles Lewis <no...@...>
Date: 2017-03-24T19:06:46Z
add task killed to exception accum test
commit cb276bc1f1583ca7dcf44b8e80a9fc2cd09953cf
Author: Charles Lewis <no...@...>
Date: 2017-03-24T23:20:59Z
extra fixes for task killed reason merge
commit 30ae1457afd9899a5fa937c2fefac31f6a0752ed
Author: Xianjin YE <ad...@...>
Date: 2018-04-26T07:34:54Z
Fix merge conflict and semantic difference
commit 88b1cebd76d7414d4bbdb99e03fe10f74e25029a
Author: Xianjin YE <ad...@...>
Date: 2018-04-26T09:22:20Z
Make accums in TaskKilled immutable and extract
common logic in TaskRunner to reduce duplicate code
----
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21165
LGTM
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r184840246
--- Diff: core/src/main/scala/org/apache/spark/TaskEndReason.scala ---
@@ -212,9 +212,15 @@ case object TaskResultLost extends TaskFailedReason {
* Task was killed intentionally and needs to be rescheduled.
*/
@DeveloperApi
-case class TaskKilled(reason: String) extends TaskFailedReason {
+case class TaskKilled(
+ reason: String,
+ accumUpdates: Seq[AccumulableInfo] = Seq.empty,
+ private[spark] val accums: Seq[AccumulatorV2[_, _]] = Nil)
--- End diff --
Yeah, I noticed `accumUpdates: Seq[AccumulableInfo]` is only used in JsonProtocol. Is that for a reason?
The current impl is constructed to be sync with existing TaskEndReason such as `ExceptionFailure`
```
@DeveloperApi
case class ExceptionFailure(
className: String,
description: String,
stackTrace: Array[StackTraceElement],
fullStackTrace: String,
private val exceptionWrapper: Option[ThrowableSerializationWrapper],
accumUpdates: Seq[AccumulableInfo] = Seq.empty,
private[spark] var accums: Seq[AccumulatorV2[_, _]] = Nil)
```
I'd prefer to keep in sync, leave two options for cleanup:
1. leave it as it is, then cleanup with ExceptionFailure together
2. Cleanup ExceptionFailure first.
@cloud-fan what do you think?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21165
**[Test build #90891 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90891/testReport)** for PR 21165 at commit [`74911b7`](https://github.com/apache/spark/commit/74911b7a8d7714618ab060b3227e33505b0c5d05).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r189530671
--- Diff: core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala ---
@@ -1868,15 +1868,26 @@ class DAGSchedulerSuite extends SparkFunSuite with LocalSparkContext with TimeLi
val accUpdate3 = new LongAccumulator
accUpdate3.metadata = acc3.metadata
accUpdate3.setValue(18)
- val accumUpdates = Seq(accUpdate1, accUpdate2, accUpdate3)
- val accumInfo = accumUpdates.map(AccumulatorSuite.makeInfo)
+
+ val accumUpdates1 = Seq(accUpdate1, accUpdate2)
+ val accumInfo1 = accumUpdates1.map(AccumulatorSuite.makeInfo)
val exceptionFailure = new ExceptionFailure(
new SparkException("fondue?"),
- accumInfo).copy(accums = accumUpdates)
+ accumInfo1).copy(accums = accumUpdates1)
submit(new MyRDD(sc, 1, Nil), Array(0))
runEvent(makeCompletionEvent(taskSets.head.tasks.head, exceptionFailure, "result"))
+
assert(AccumulatorContext.get(acc1.id).get.value === 15L)
assert(AccumulatorContext.get(acc2.id).get.value === 13L)
+
+ val accumUpdates2 = Seq(accUpdate3)
+ val accumInfo2 = accumUpdates2.map(AccumulatorSuite.makeInfo)
+
+ val taskKilled = new TaskKilled(
+ "test",
+ accumInfo2).copy(accums = accumUpdates2)
--- End diff --
We can avoid this `copy` call
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r189525864
--- Diff: core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala ---
@@ -1868,15 +1868,26 @@ class DAGSchedulerSuite extends SparkFunSuite with LocalSparkContext with TimeLi
val accUpdate3 = new LongAccumulator
accUpdate3.metadata = acc3.metadata
accUpdate3.setValue(18)
- val accumUpdates = Seq(accUpdate1, accUpdate2, accUpdate3)
- val accumInfo = accumUpdates.map(AccumulatorSuite.makeInfo)
+
+ val accumUpdates1 = Seq(accUpdate1, accUpdate2)
+ val accumInfo1 = accumUpdates1.map(AccumulatorSuite.makeInfo)
val exceptionFailure = new ExceptionFailure(
new SparkException("fondue?"),
- accumInfo).copy(accums = accumUpdates)
+ accumInfo1).copy(accums = accumUpdates1)
--- End diff --
We can avoid the `copy` call.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r187987691
--- Diff: core/src/main/scala/org/apache/spark/executor/Executor.scala ---
@@ -287,6 +287,28 @@ private[spark] class Executor(
notifyAll()
}
+ /**
+ * Utility function to:
+ * 1. Report executor runtime and JVM gc time if possible
+ * 2. Collect accumulator updates
+ * 3. Set the finished flag to true and clear current thread's interrupt status
+ */
+ private def collectAccumulatorsAndResetStatusOnFailure(taskStart: Long) = {
+ // Report executor runtime and JVM gc time
+ Option(task).foreach(t => {
+ t.metrics.setExecutorRunTime(System.currentTimeMillis() - taskStart)
--- End diff --
we should at least rename it in this method as it's newly added code. We can also update the existing code if it's not a lot of work.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r189530510
--- Diff: core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala ---
@@ -1868,15 +1868,26 @@ class DAGSchedulerSuite extends SparkFunSuite with LocalSparkContext with TimeLi
val accUpdate3 = new LongAccumulator
accUpdate3.metadata = acc3.metadata
accUpdate3.setValue(18)
- val accumUpdates = Seq(accUpdate1, accUpdate2, accUpdate3)
- val accumInfo = accumUpdates.map(AccumulatorSuite.makeInfo)
+
+ val accumUpdates1 = Seq(accUpdate1, accUpdate2)
+ val accumInfo1 = accumUpdates1.map(AccumulatorSuite.makeInfo)
val exceptionFailure = new ExceptionFailure(
new SparkException("fondue?"),
- accumInfo).copy(accums = accumUpdates)
+ accumInfo1).copy(accums = accumUpdates1)
--- End diff --
Ah, this `copy` call cannot be avoided as only the 2 arguments constructor
``` private[spark] def this(e: Throwable, accumUpdates: Seq[AccumulableInfo]) ``` is defined.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: Spark 20087: Attach accumulators / metrics to 'Ta...
Posted by squito <gi...@git.apache.org>.
Github user squito commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r184404291
--- Diff: core/src/main/scala/org/apache/spark/executor/Executor.scala ---
@@ -287,6 +287,33 @@ private[spark] class Executor(
notifyAll()
}
+ /**
+ * Set executor runtime and JVM gc time if task instance is still valid
+ */
+ private def reportGCAndExecutorTimeIfPossible(taskStart: Long): Unit = {
+ if (task != null) {
+ task.metrics.setExecutorRunTime(System.currentTimeMillis() - taskStart)
+ task.metrics.setJvmGCTime(computeTotalGcTime() - startGCTime)
+ }
+ }
+
+ /**
+ * Utility function to:
+ * 1. Report executor runtime and JVM gc time if possible
+ * 2. Collect accumulator updates
+ * 3. Set the finished flag to true and clear current thread's interrupt status
+ */
+ private def collectAccumulatorsAndResetStatusOnFailure(taskStart: Long) = {
+ reportGCAndExecutorTimeIfPossible(taskStart)
--- End diff --
I don't think the extra `reportGCAndExecutorTimeIfPossible` is necessary, you can just inline it. and also the original `if (task != null)` is probably easier to follow than `Option(task).map`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on the issue:
https://github.com/apache/spark/pull/21165
@jiangxb1987 @cloud-fan I think it's ready for review.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on the issue:
https://github.com/apache/spark/pull/21165
> We should not do these 2 things together, and to me the second one is way simpler to get in and we should do it first.
Agreed. For the scope of this pr, let's get killed tasks's accumulators into metrics first. After that we can discuss the possibility to expose the ability under users' request.
> but please make sure this patch only touches internal accumulators that are used for metrics reporting.
After a second look, this part is already be handled by Task's collectAccumulatorUpdates:
```
def collectAccumulatorUpdates(taskFailed: Boolean = false): Seq[AccumulatorV2[_, _]] = {
if (context != null) {
// Note: internal accumulators representing task metrics always count failed values
context.taskMetrics.nonZeroInternalAccums() ++
// zero value external accumulators may still be useful, e.g. SQLMetrics, we should not
// filter them out.
context.taskMetrics.externalAccums.filter(a => !taskFailed || a.countFailedValues)
} else {
Seq.empty
}
}
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21165
**[Test build #90643 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90643/testReport)** for PR 21165 at commit [`59c2807`](https://github.com/apache/spark/commit/59c2807dca5cc1c9332a1b3e7fcc73214df436ec).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21165
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r186468071
--- Diff: core/src/main/scala/org/apache/spark/TaskEndReason.scala ---
@@ -212,9 +212,15 @@ case object TaskResultLost extends TaskFailedReason {
* Task was killed intentionally and needs to be rescheduled.
*/
@DeveloperApi
-case class TaskKilled(reason: String) extends TaskFailedReason {
+case class TaskKilled(
+ reason: String,
+ accumUpdates: Seq[AccumulableInfo] = Seq.empty,
+ private[spark] val accums: Seq[AccumulatorV2[_, _]] = Nil)
--- End diff --
Hi @cloud-fan, I have looked at how to remove `Seq[AccumulableInfo]` tonight.
It turns out that we cannot because `JsonProtocol` calls `taskEndReasonFromJson` to reconstruct `TaskEndReason`s. Since `AccumulatorV2` is an abstract class, we cannot simply construct `AccumulatorV2`s from json.
Even we are promoting `AccumulatorV2`, we still need `AccumulableInfo` when (de)serializing json.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21165
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r188005420
--- Diff: core/src/main/scala/org/apache/spark/executor/Executor.scala ---
@@ -287,6 +287,28 @@ private[spark] class Executor(
notifyAll()
}
+ /**
+ * Utility function to:
+ * 1. Report executor runtime and JVM gc time if possible
+ * 2. Collect accumulator updates
+ * 3. Set the finished flag to true and clear current thread's interrupt status
+ */
+ private def collectAccumulatorsAndResetStatusOnFailure(taskStart: Long) = {
+ // Report executor runtime and JVM gc time
+ Option(task).foreach(t => {
+ t.metrics.setExecutorRunTime(System.currentTimeMillis() - taskStart)
--- End diff --
Will do it..
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21165
**[Test build #90585 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90585/testReport)** for PR 21165 at commit [`05d1d9c`](https://github.com/apache/spark/commit/05d1d9cad761bb09e1131162458fecd5e34f02d2).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r187905421
--- Diff: docs/rdd-programming-guide.md ---
@@ -1548,6 +1548,9 @@ data.map(g)
</div>
+In new version of Spark(> 2.3), the semantic of Accumulator has been changed a bit: it now includes updates from
--- End diff --
it's not needed now
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on the issue:
https://github.com/apache/spark/pull/21165
I add a note for accumulator update. Please comment if more document is needed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21165
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/90901/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21165
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r185226136
--- Diff: core/src/main/scala/org/apache/spark/TaskEndReason.scala ---
@@ -212,9 +212,15 @@ case object TaskResultLost extends TaskFailedReason {
* Task was killed intentionally and needs to be rescheduled.
*/
@DeveloperApi
-case class TaskKilled(reason: String) extends TaskFailedReason {
+case class TaskKilled(
+ reason: String,
+ accumUpdates: Seq[AccumulableInfo] = Seq.empty,
+ private[spark] val accums: Seq[AccumulatorV2[_, _]] = Nil)
--- End diff --
@cloud-fan After a second look, I don't think we can clean up `ExceptionFailure` unless we can break `JsonProtocol`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21165
ok to test
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21165
**[Test build #90643 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90643/testReport)** for PR 21165 at commit [`59c2807`](https://github.com/apache/spark/commit/59c2807dca5cc1c9332a1b3e7fcc73214df436ec).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21165
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r187984186
--- Diff: core/src/main/scala/org/apache/spark/executor/Executor.scala ---
@@ -287,6 +287,28 @@ private[spark] class Executor(
notifyAll()
}
+ /**
+ * Utility function to:
+ * 1. Report executor runtime and JVM gc time if possible
+ * 2. Collect accumulator updates
+ * 3. Set the finished flag to true and clear current thread's interrupt status
+ */
+ private def collectAccumulatorsAndResetStatusOnFailure(taskStart: Long) = {
+ // Report executor runtime and JVM gc time
+ Option(task).foreach(t => {
+ t.metrics.setExecutorRunTime(System.currentTimeMillis() - taskStart)
--- End diff --
e, `startStart` is already defined previously. Do you think we need to replace all the `taskStart` to `taskStartTime`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r188925889
--- Diff: core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala ---
@@ -1868,15 +1868,26 @@ class DAGSchedulerSuite extends SparkFunSuite with LocalSparkContext with TimeLi
val accUpdate3 = new LongAccumulator
accUpdate3.metadata = acc3.metadata
accUpdate3.setValue(18)
- val accumUpdates = Seq(accUpdate1, accUpdate2, accUpdate3)
- val accumInfo = accumUpdates.map(AccumulatorSuite.makeInfo)
+
+ val accumUpdates1 = Seq(accUpdate1, accUpdate2)
+ val accumInfo1 = accumUpdates1.map(AccumulatorSuite.makeInfo)
val exceptionFailure = new ExceptionFailure(
new SparkException("fondue?"),
- accumInfo).copy(accums = accumUpdates)
+ accumInfo1).copy(accums = accumUpdates1)
--- End diff --
not caused by you but why we do a copy instead of passing `accumUpdates1` to the constructor directly?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: Spark 20087: Attach accumulators / metrics to 'TaskKille...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on the issue:
https://github.com/apache/spark/pull/21165
cc @squito @jiangxb1987
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on the issue:
https://github.com/apache/spark/pull/21165
ping @cloud-fan
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21165
> For example user may want to record CPU time for every task and get the total CPU time for the application.
The problem is, shall we allow end users to collect metrics via accumulators? Currently only Spark can do that via internal accumulators which count failed tasks. We need a careful API design about how to expose this ability in the end users.
In the meanwhile, since we already count failed tasks, it makes sense to also count killed tasks for internal metrics collecting.
We should not do these 2 things together, and to me the second one is way simpler to get in and we should do it first.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark 20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on the issue:
https://github.com/apache/spark/pull/21165
> we should document the changes in a migration document or something,
I think documentation is necessary, will update the documentation tomorrow (Beijing time)
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: Spark 20087: Attach accumulators / metrics to 'TaskKille...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21165
Can one of the admins verify this patch?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r186930874
--- Diff: core/src/main/scala/org/apache/spark/TaskEndReason.scala ---
@@ -212,9 +212,15 @@ case object TaskResultLost extends TaskFailedReason {
* Task was killed intentionally and needs to be rescheduled.
*/
@DeveloperApi
-case class TaskKilled(reason: String) extends TaskFailedReason {
+case class TaskKilled(
+ reason: String,
+ accumUpdates: Seq[AccumulableInfo] = Seq.empty,
+ private[spark] val accums: Seq[AccumulatorV2[_, _]] = Nil)
--- End diff --
I see, that makes sense, let's keep `AccumulableInfo`.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on the issue:
https://github.com/apache/spark/pull/21165
Looks like that simply add fields with default values into case class will break binary compatibility.
How should we deal with that? Add to MimaExcludes or add missing methods? @cloud-fan
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21165
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/90891/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21165
**[Test build #90585 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90585/testReport)** for PR 21165 at commit [`05d1d9c`](https://github.com/apache/spark/commit/05d1d9cad761bb09e1131162458fecd5e34f02d2).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21165
**[Test build #90901 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90901/testReport)** for PR 21165 at commit [`74911b7`](https://github.com/apache/spark/commit/74911b7a8d7714618ab060b3227e33505b0c5d05).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21165
thanks, merging to master!
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on the issue:
https://github.com/apache/spark/pull/21165
Gently ping @cloud-fan again.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21165
**[Test build #90891 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90891/testReport)** for PR 21165 at commit [`74911b7`](https://github.com/apache/spark/commit/74911b7a8d7714618ab060b3227e33505b0c5d05).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21165
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21165
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/90643/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on the issue:
https://github.com/apache/spark/pull/21165
> However, I don't agree user side accumulators should get updates from killed tasks, that changes the semantic of accumulators. And I don't think end-users need to care about killed tasks. Similarly, when we implement task metrics, we need to count failed tasks, but user side accumulator still skips failed tasks. I think we should also follow that approach.
I don't agree that end-user didn't care killed tasks. For example user may want to record CPU time for every task and get the total CPU time for the application. However the default behaviour should keep backward-compatibility with existing behaviour.
```
private[spark] case class AccumulatorMetadata(
id: Long,
name: Option[String],
countFailedValues: Boolean) extends Serializable
```
The metadata has `countFailedValues` field, we can use this or add a new field?
However we didn't expose this field to end user...
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21165
LGTM
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21165
**[Test build #90901 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90901/testReport)** for PR 21165 at commit [`74911b7`](https://github.com/apache/spark/commit/74911b7a8d7714618ab060b3227e33505b0c5d05).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark 20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by jiangxb1987 <gi...@git.apache.org>.
Github user jiangxb1987 commented on the issue:
https://github.com/apache/spark/pull/21165
It should be `[Spark-20087]` instead of `[Spark 20087]` in the title.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21165
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/90585/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: Spark 20087: Attach accumulators / metrics to 'Ta...
Posted by jiangxb1987 <gi...@git.apache.org>.
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r184436634
--- Diff: core/src/main/scala/org/apache/spark/TaskEndReason.scala ---
@@ -212,9 +212,15 @@ case object TaskResultLost extends TaskFailedReason {
* Task was killed intentionally and needs to be rescheduled.
*/
@DeveloperApi
-case class TaskKilled(reason: String) extends TaskFailedReason {
- override def toErrorString: String = s"TaskKilled ($reason)"
+case class TaskKilled(
+ reason: String,
+ accumUpdates: Seq[AccumulableInfo] = Seq.empty,
+ private[spark] val accums: Seq[AccumulatorV2[_, _]] = Nil)
+ extends TaskFailedReason {
+
+ override def toErrorString: String = "TaskKilled ($reason)"
--- End diff --
`s"TaskKilled ($reason)"`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r184841269
--- Diff: core/src/main/scala/org/apache/spark/TaskEndReason.scala ---
@@ -212,9 +212,15 @@ case object TaskResultLost extends TaskFailedReason {
* Task was killed intentionally and needs to be rescheduled.
*/
@DeveloperApi
-case class TaskKilled(reason: String) extends TaskFailedReason {
+case class TaskKilled(
+ reason: String,
+ accumUpdates: Seq[AccumulableInfo] = Seq.empty,
+ private[spark] val accums: Seq[AccumulatorV2[_, _]] = Nil)
--- End diff --
let's clean up ExceptionFailure at the same time, and use only `AccumulatorV2` in this PR.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r188925916
--- Diff: core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala ---
@@ -1868,15 +1868,26 @@ class DAGSchedulerSuite extends SparkFunSuite with LocalSparkContext with TimeLi
val accUpdate3 = new LongAccumulator
accUpdate3.metadata = acc3.metadata
accUpdate3.setValue(18)
- val accumUpdates = Seq(accUpdate1, accUpdate2, accUpdate3)
- val accumInfo = accumUpdates.map(AccumulatorSuite.makeInfo)
+
+ val accumUpdates1 = Seq(accUpdate1, accUpdate2)
+ val accumInfo1 = accumUpdates1.map(AccumulatorSuite.makeInfo)
val exceptionFailure = new ExceptionFailure(
new SparkException("fondue?"),
- accumInfo).copy(accums = accumUpdates)
+ accumInfo1).copy(accums = accumUpdates1)
submit(new MyRDD(sc, 1, Nil), Array(0))
runEvent(makeCompletionEvent(taskSets.head.tasks.head, exceptionFailure, "result"))
+
assert(AccumulatorContext.get(acc1.id).get.value === 15L)
assert(AccumulatorContext.get(acc2.id).get.value === 13L)
+
+ val accumUpdates2 = Seq(accUpdate3)
+ val accumInfo2 = accumUpdates2.map(AccumulatorSuite.makeInfo)
+
+ val taskKilled = new TaskKilled(
+ "test",
+ accumInfo2).copy(accums = accumUpdates2)
--- End diff --
ditto
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r185688449
--- Diff: core/src/main/scala/org/apache/spark/TaskEndReason.scala ---
@@ -212,9 +212,15 @@ case object TaskResultLost extends TaskFailedReason {
* Task was killed intentionally and needs to be rescheduled.
*/
@DeveloperApi
-case class TaskKilled(reason: String) extends TaskFailedReason {
+case class TaskKilled(
+ reason: String,
+ accumUpdates: Seq[AccumulableInfo] = Seq.empty,
+ private[spark] val accums: Seq[AccumulatorV2[_, _]] = Nil)
--- End diff --
now the question is: shall we keep the unnecessary Seq[AccumulableInfo] in new APIs, to make the API consistent? I'd like to not keep the Seq[AccumulableInfo], we may deprecate it in the existing APIs in the near future.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: Spark 20087: Attach accumulators / metrics to 'Ta...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r184440128
--- Diff: core/src/main/scala/org/apache/spark/TaskEndReason.scala ---
@@ -212,9 +212,15 @@ case object TaskResultLost extends TaskFailedReason {
* Task was killed intentionally and needs to be rescheduled.
*/
@DeveloperApi
-case class TaskKilled(reason: String) extends TaskFailedReason {
- override def toErrorString: String = s"TaskKilled ($reason)"
+case class TaskKilled(
+ reason: String,
+ accumUpdates: Seq[AccumulableInfo] = Seq.empty,
+ private[spark] val accums: Seq[AccumulatorV2[_, _]] = Nil)
+ extends TaskFailedReason {
+
+ override def toErrorString: String = "TaskKilled ($reason)"
--- End diff --
Will do.
Didn't notice `s` is missing
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: Spark 20087: Attach accumulators / metrics to 'Ta...
Posted by jiangxb1987 <gi...@git.apache.org>.
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r184438550
--- Diff: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala ---
@@ -1417,10 +1417,13 @@ class DAGScheduler(
case exceptionFailure: ExceptionFailure =>
// Nothing left to do, already handled above for accumulator updates.
+ case _: TaskKilled =>
--- End diff --
nit: combine this with the `ExceptionFailure` case.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21165
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/90595/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: Spark 20087: Attach accumulators / metrics to 'TaskKille...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21165
Can one of the admins verify this patch?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r184838512
--- Diff: core/src/main/scala/org/apache/spark/TaskEndReason.scala ---
@@ -212,9 +212,15 @@ case object TaskResultLost extends TaskFailedReason {
* Task was killed intentionally and needs to be rescheduled.
*/
@DeveloperApi
-case class TaskKilled(reason: String) extends TaskFailedReason {
+case class TaskKilled(
+ reason: String,
+ accumUpdates: Seq[AccumulableInfo] = Seq.empty,
+ private[spark] val accums: Seq[AccumulatorV2[_, _]] = Nil)
--- End diff --
Previously we use `AccumulableInfo` to expose accumulator information to end users. Now `AccumulatorV2` is already a public classs and we don't need to do it anymore, I think we can just do
```
case class TaskKilled(reason: String, accums: Seq[AccumulatorV2[_, _]])
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r185690094
--- Diff: core/src/main/scala/org/apache/spark/TaskEndReason.scala ---
@@ -212,9 +212,15 @@ case object TaskResultLost extends TaskFailedReason {
* Task was killed intentionally and needs to be rescheduled.
*/
@DeveloperApi
-case class TaskKilled(reason: String) extends TaskFailedReason {
+case class TaskKilled(
+ reason: String,
+ accumUpdates: Seq[AccumulableInfo] = Seq.empty,
+ private[spark] val accums: Seq[AccumulatorV2[_, _]] = Nil)
--- End diff --
I'm ok with not keeping `Seq[AccumulableInfo]`. But it means inconsistent logic and api and may make future refactoring a bit difficult.
Let's see what I can do.
> I'd like to not keep the Seq[AccumulableInfo], we may deprecate it in the existing APIs in the near future.
BTW, I think we have already deprecated `AccumulableInfo`. Unless we are planing to remove it in Spark 3.0 and Spark 3.0 is the next release, `AccumulableInfo` will be there for a long time
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: Spark 20087: Attach accumulators / metrics to 'TaskKille...
Posted by squito <gi...@git.apache.org>.
Github user squito commented on the issue:
https://github.com/apache/spark/pull/21165
Ok to test
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/21165
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r187074691
--- Diff: core/src/main/scala/org/apache/spark/TaskEndReason.scala ---
@@ -212,9 +212,15 @@ case object TaskResultLost extends TaskFailedReason {
* Task was killed intentionally and needs to be rescheduled.
*/
@DeveloperApi
-case class TaskKilled(reason: String) extends TaskFailedReason {
+case class TaskKilled(
+ reason: String,
+ accumUpdates: Seq[AccumulableInfo] = Seq.empty,
+ private[spark] val accums: Seq[AccumulatorV2[_, _]] = Nil)
--- End diff --
@cloud-fan so, could you trigger the test and have a look?
And looks like I am not in the whitelist again...
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21165
**[Test build #90595 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90595/testReport)** for PR 21165 at commit [`945c1d5`](https://github.com/apache/spark/commit/945c1d5945e2cf836e36f4c670b5b80dc83b1c76).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21165
I do agree task killed event should carry metrics update, as it's reasonable to count killed tasks for something like how many bytes were read from files.
However, I don't agree user side accumulators should get updates from killed tasks, that changes the semantic of accumulators. And I don't think end-users need to care about killed tasks.
I haven't read the PR yet, but please make sure this patch only touches internal accumulators that are used for metrics reporting.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: Spark 20087: Attach accumulators / metrics to 'TaskKille...
Posted by jiangxb1987 <gi...@git.apache.org>.
Github user jiangxb1987 commented on the issue:
https://github.com/apache/spark/pull/21165
I'm not against the change, but since this changes the semantics of accumulators, we should document the changes in a migration document or something, WDYT @cloud-fan @gatorsmile ?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: Spark 20087: Attach accumulators / metrics to 'Ta...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r184429082
--- Diff: core/src/main/scala/org/apache/spark/executor/Executor.scala ---
@@ -287,6 +287,28 @@ private[spark] class Executor(
notifyAll()
}
+ /**
+ * Utility function to:
+ * 1. Report executor runtime and JVM gc time if possible
+ * 2. Collect accumulator updates
+ * 3. Set the finished flag to true and clear current thread's interrupt status
+ */
+ private def collectAccumulatorsAndResetStatusOnFailure(taskStart: Long) = {
--- End diff --
@squito after address your comment, do you think we should come up with a more specific method name?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: Spark 20087: Attach accumulators / metrics to 'Ta...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r184441076
--- Diff: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala ---
@@ -1417,10 +1417,13 @@ class DAGScheduler(
case exceptionFailure: ExceptionFailure =>
// Nothing left to do, already handled above for accumulator updates.
+ case _: TaskKilled =>
--- End diff --
will do.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on the issue:
https://github.com/apache/spark/pull/21165
> It should be [Spark-20087] instead of [Spark 20087] in the title.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by advancedxy <gi...@git.apache.org>.
Github user advancedxy commented on the issue:
https://github.com/apache/spark/pull/21165
retest this please
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: Spark 20087: Attach accumulators / metrics to 'TaskKille...
Posted by jiangxb1987 <gi...@git.apache.org>.
Github user jiangxb1987 commented on the issue:
https://github.com/apache/spark/pull/21165
Also, please update the PR title:
`[Spark-20087][CORE] Attach accumulators / metrics to 'TaskKilled' end reason`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21165
retest this please
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21165
**[Test build #90595 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90595/testReport)** for PR 21165 at commit [`945c1d5`](https://github.com/apache/spark/commit/945c1d5945e2cf836e36f4c670b5b80dc83b1c76).
* This patch **fails MiMa tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21165: [Spark-20087][CORE] Attach accumulators / metrics to 'Ta...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21165
I think we can just update MimaExcludes, since it's developer API. cc @JoshRosen
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21165: [Spark-20087][CORE] Attach accumulators / metrics...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21165#discussion_r187981280
--- Diff: core/src/main/scala/org/apache/spark/executor/Executor.scala ---
@@ -287,6 +287,28 @@ private[spark] class Executor(
notifyAll()
}
+ /**
+ * Utility function to:
+ * 1. Report executor runtime and JVM gc time if possible
+ * 2. Collect accumulator updates
+ * 3. Set the finished flag to true and clear current thread's interrupt status
+ */
+ private def collectAccumulatorsAndResetStatusOnFailure(taskStart: Long) = {
+ // Report executor runtime and JVM gc time
+ Option(task).foreach(t => {
+ t.metrics.setExecutorRunTime(System.currentTimeMillis() - taskStart)
--- End diff --
`taskStartTime`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org