[SPARK-22897][CORE]: Expose stageAttemptId in TaskContext #20082

advancedxy · 2017-12-26T11:59:53Z

What changes were proposed in this pull request?

stageAttemptId added in TaskContext and corresponding construction modification

How was this patch tested?

Added a new test in TaskContextSuite, two cases are tested:

Normal case without failure
Exception case with resubmitted stages

Link to SPARK-22897

advancedxy · 2017-12-26T12:00:54Z

cc @cloud-fan

cloud-fan · 2017-12-26T12:03:28Z

core/src/main/scala/org/apache/spark/TaskContext.scala

@@ -150,6 +150,10 @@ abstract class TaskContext extends Serializable {
   */
  def stageId(): Int

+  /**
+   * The attempt ID of the stage that this task belongs to.


can we follow taskAttemptId and say more in the document?

Of course, I can add more doc.

But is it necessary? I think the field(function) name is already self-explanatory.

jiangxb1987 · 2017-12-26T14:32:00Z

Please explain in the JIRA/PR description about your use case that need to access the stageAttemptId in the executor - we must provide a very good reason to make such a change to a public interface.

cloud-fan · 2017-12-26T16:05:51Z

ok to test

SparkQA · 2017-12-26T16:19:02Z

Test build #85409 has finished for PR 20082 at commit 5753ee0.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds no public classes.

advancedxy · 2017-12-26T16:25:35Z

Please explain in the JIRA/PR description about your use case that need to access the stageAttemptId in the executor - we must provide a very good reason to make such a change to a public interface.

@jiangxb1987 quoted from SPARK-22897:

Currently, there's no easy way for Executor to detect a new stage is launched as stageAttemptId is missing.

In one of our use case, spark task needs to reset some internal state once the stage is changed(we don't want to reset the state for every task as there may be a lot of tasks).

IMHO, stageAttemptId is as important as stageId or taskAttemptId and should be provided to user.

As for compatibility, we should do our best to avoid any incompatibility in minor releases. However,
this change should be source and binary compatible with previous releases.

1. Update mima exclude 2. Update wording in TaskContext 3. Update JavaTaskContextCompileCheck as suggested in TaskContext note

advancedxy · 2017-12-26T16:45:09Z

project/MimaExcludes.scala

+    ProblemFilters.exclude[FinalClassProblem]("org.apache.spark.ml.tuning.TrainValidationSplitModel$TrainValidationSplitModelWriter"),
+
+    // [SPARK-22897] Expose stageAttemptId in TaskContext
+    ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.TaskContext.stageAttemptId")


This change is suggested by #12248

If not appropriate, please let me know.

can you put it at the beginning of v23excludes?

SparkQA · 2017-12-26T19:48:50Z

Test build #85410 has finished for PR 20082 at commit f02bc1e.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-12-27T02:43:51Z

Test build #85420 has finished for PR 20082 at commit 0f86e95.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-12-27T06:18:45Z

Test build #85421 has finished for PR 20082 at commit 59e4a9c.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

jiangxb1987 · 2017-12-27T07:18:18Z

core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala

+
+    // Check stage attemptIds that are resubmitted when task fails
+    val stageAttemptIdsWithFailedStage =
+      sc.parallelize(Seq(1, 2, 3, 4), 4).repartition(1).mapPartitions { _ =>


You don't need repartition here, just sc.parallelize(Seq(1, 2, 3, 4), 1).mapPartitions {...}

jiangxb1987 · 2017-12-27T07:20:21Z

core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala

+      sc.parallelize(Seq(1, 2, 3, 4), 4).repartition(1).mapPartitions { _ =>
+      val stageAttemptId = TaskContext.get().stageAttemptId()
+      if (stageAttemptId < 2) {
+        throw new FetchFailedException(null, 0, 0, 0, "Fake")


Emmm... just throw an Exception is enough here?

Related to repartition part.

I use FetchFailedException to explicitly trigger a stage resubmission. Otherwise, the task would be resubmitted in the same stage if IIRC.

Please add comment to explain that FetchFailedException will trigger a new stage attempt, while a common Exception will only trigger a task retry.

SparkQA · 2017-12-27T11:52:47Z

Test build #85429 has finished for PR 20082 at commit 291bbbc.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing

LGTM except one nit

zsxwing · 2017-12-27T18:59:25Z

core/src/main/scala/org/apache/spark/TaskContextImpl.scala

@@ -42,6 +42,7 @@ import org.apache.spark.util._
 */
 private[spark] class TaskContextImpl(
    val stageId: Int,
+    val stageAttemptId: Int,


nit: add override. Since you are touching this file, could you also add override to stageId and partitionId.

Will do.

Would you tell me the difference or rationale?

it's kind of a code style standard: add override if it is override.

cloud-fan · 2017-12-28T04:57:53Z

core/src/main/scala/org/apache/spark/TaskContext.scala

+  /**
+   * An ID that is unique to the stage attempt that this task belongs to.
+   */
+  def stageAttemptId(): Int


I think we should call it stageAttempNumber to be consistent with taskAttemptNumber. Also let's follow the comment of attemptNumber

Yeah, if we are defining stageAttemptId from scratch, I would go for stageAttemptNumber. However stageAttemptId are already used elsewhere in the codebase, Like in Task.scala. I think it's more important to be consistent.

However I could update the comment to reflect the attempt number part if you wish

Updating wording in TaskContext.stageAttemptId

SparkQA · 2017-12-28T08:05:01Z

Test build #85457 has finished for PR 20082 at commit 72a3abf.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

jiangxb1987 · 2017-12-28T14:13:08Z

retest this please

SparkQA · 2017-12-28T17:52:33Z

Test build #85474 has finished for PR 20082 at commit 72a3abf.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

advancedxy · 2017-12-29T04:02:55Z

ping @cloud-fan @jiangxb1987 @zsxwing, I think it's ready for merging.

cloud-fan · 2017-12-29T07:51:31Z

core/src/main/scala/org/apache/spark/TaskContext.scala

+   * times the stage has been attempted. The first stage attempt will be assigned stageAttemptId = 0
+   * , and subsequent attempts will increasing stageAttemptId one by one.
+   */
+  def stageAttemptId(): Int


My concern is that, internally we use stageAttemptId, and internally we call TaskContext.taskAttemptId taskId. However, for end users, they don't know the internal code, and they are more familiar with TaskContext. I think the naming should be consistent with the public API TaskContext, instead of internal code.

I have no objection for either ‘id’ nor ‘number’, they are both reasonable.

I am on train now。 If no other input, I can rename it to ‘stageAttemptNumber’ since you insisted.

SparkQA · 2017-12-29T18:48:47Z

Test build #85512 has finished for PR 20082 at commit 2581263.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-12-29T18:56:36Z

Test build #85513 has finished for PR 20082 at commit 9266cd8.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

advancedxy · 2018-01-02T02:41:09Z

@cloud-fan Please take another look.

cloud-fan · 2018-01-02T05:19:49Z

core/src/main/scala/org/apache/spark/scheduler/Task.scala

@@ -79,6 +79,7 @@ private[spark] abstract class Task[T](
    SparkEnv.get.blockManager.registerTask(taskAttemptId)
    context = new TaskContextImpl(
      stageId,
+      stageAttemptId, // stageAttemptId and stageAttemptNumber are semantically equal


How much work we need to rename the internal stageAttemptId to stageAttemptNumber?

The modification may not be too much (100+ occurrences in 20+ files), however it may break eventLog's JsonProtocol backward compatibility(not sure)..

@squito you may have more knowledge on this since you introduced stageAttemptId.

ah, so stageAttemptId is already exposed in developer API, we can't change it.

cloud-fan · 2018-01-02T05:20:08Z

LGTM

## What changes were proposed in this pull request? stageAttemptId added in TaskContext and corresponding construction modification ## How was this patch tested? Added a new test in TaskContextSuite, two cases are tested: 1. Normal case without failure 2. Exception case with resubmitted stages Link to [SPARK-22897](https://issues.apache.org/jira/browse/SPARK-22897) Author: Xianjin YE <advancedxy@gmail.com> Closes #20082 from advancedxy/SPARK-22897. (cherry picked from commit a6fc300) Signed-off-by: Wenchen Fan <wenchen@databricks.com>

cloud-fan · 2018-01-02T15:31:24Z

thanks, merging to master/2.3!

zsxwing · 2018-01-03T20:55:55Z

@advancedxy could you also submit a PR to add a def stageAttemptNumber = stageAttemptId to StageInfo and mark stageAttemptId as deprecated? It will make our public APIs more consistent.

advancedxy · 2018-01-04T00:52:07Z

Of course.

squito · 2018-01-05T05:48:30Z

sorry this is a little late, but lgtm too. agree with the points above about leaving the old name deprecated and moving to the new name

stageAttemptId added in TaskContext and corresponding construction modification Added a new test in TaskContextSuite, two cases are tested: 1. Normal case without failure 2. Exception case with resubmitted stages Link to [SPARK-22897](https://issues.apache.org/jira/browse/SPARK-22897) Author: Xianjin YE <advancedxy@gmail.com> Closes apache#20082 from advancedxy/SPARK-22897. (cherry picked from commit a6fc300) Signed-off-by: Marcelo Vanzin <vanzin@cloudera.com>

stageAttemptId added in TaskContext and corresponding construction modification Added a new test in TaskContextSuite, two cases are tested: 1. Normal case without failure 2. Exception case with resubmitted stages Link to [SPARK-22897](https://issues.apache.org/jira/browse/SPARK-22897) Author: Xianjin YE <advancedxy@gmail.com> Closes apache#20082 from advancedxy/SPARK-22897. Conflicts: project/MimaExcludes.scala

stageAttemptId added in TaskContext and corresponding construction modification Added a new test in TaskContextSuite, two cases are tested: 1. Normal case without failure 2. Exception case with resubmitted stages Link to [SPARK-22897](https://issues.apache.org/jira/browse/SPARK-22897) Author: Xianjin YE <advancedxygmail.com> Closes #20082 from advancedxy/SPARK-22897. Conflicts: project/MimaExcludes.scala ## What changes were proposed in this pull request? (Please fill in changes proposed in this fix) ## How was this patch tested? (Please explain how this patch was tested. E.g. unit tests, integration tests, manual tests) (If this patch involves UI changes, please attach a screenshot; otherwise, remove this) Please review http://spark.apache.org/contributing.html before opening a pull request. Author: Xianjin YE <advancedxy@gmail.com> Author: Thomas Graves <tgraves@unharmedunarmed.corp.ne1.yahoo.com> Closes #21609 from tgravescs/SPARK-22897.

stageAttemptId added in TaskContext and corresponding construction modification Added a new test in TaskContextSuite, two cases are tested: 1. Normal case without failure 2. Exception case with resubmitted stages Link to [SPARK-22897](https://issues.apache.org/jira/browse/SPARK-22897) Author: Xianjin YE <advancedxygmail.com> Closes apache#20082 from advancedxy/SPARK-22897. Conflicts: project/MimaExcludes.scala ## What changes were proposed in this pull request? (Please fill in changes proposed in this fix) ## How was this patch tested? (Please explain how this patch was tested. E.g. unit tests, integration tests, manual tests) (If this patch involves UI changes, please attach a screenshot; otherwise, remove this) Please review http://spark.apache.org/contributing.html before opening a pull request. Author: Xianjin YE <advancedxy@gmail.com> Author: Thomas Graves <tgraves@unharmedunarmed.corp.ne1.yahoo.com> Closes apache#21609 from tgravescs/SPARK-22897.

stageAttemptId added in TaskContext and corresponding construction modification Added a new test in TaskContextSuite, two cases are tested: 1. Normal case without failure 2. Exception case with resubmitted stages Link to [SPARK-22897](https://issues.apache.org/jira/browse/SPARK-22897) Author: Xianjin YE <advancedxy@gmail.com> Closes apache#20082 from advancedxy/SPARK-22897. (cherry picked from commit a6fc300) Signed-off-by: Marcelo Vanzin <vanzin@cloudera.com>

stageAttemptId added in TaskContext and corresponding construction modification Added a new test in TaskContextSuite, two cases are tested: 1. Normal case without failure 2. Exception case with resubmitted stages Link to [SPARK-22897](https://issues.apache.org/jira/browse/SPARK-22897) Author: Xianjin YE <advancedxygmail.com> Closes apache#20082 from advancedxy/SPARK-22897. Conflicts: project/MimaExcludes.scala (Please fill in changes proposed in this fix) (Please explain how this patch was tested. E.g. unit tests, integration tests, manual tests) (If this patch involves UI changes, please attach a screenshot; otherwise, remove this) Please review http://spark.apache.org/contributing.html before opening a pull request. Author: Xianjin YE <advancedxy@gmail.com> Author: Thomas Graves <tgraves@unharmedunarmed.corp.ne1.yahoo.com> Closes apache#21609 from tgravescs/SPARK-22897.

[SPARK-22897][CORE]: Expose stageAttemptId in TaskContext

5753ee0

cloud-fan reviewed Dec 26, 2017

View reviewed changes

Minor fixes such as:

f02bc1e

1. Update mima exclude 2. Update wording in TaskContext 3. Update JavaTaskContextCompileCheck as suggested in TaskContext note

advancedxy commented Dec 26, 2017

View reviewed changes

Update MimaExcludes to move stageAttemptId filter to the beginning

59e4a9c

advancedxy force-pushed the SPARK-22897 branch from 0f86e95 to 59e4a9c Compare December 27, 2017 02:46

jiangxb1987 reviewed Dec 27, 2017

View reviewed changes

Add comment for usage of FetchFailedException

291bbbc

zsxwing reviewed Dec 27, 2017

View reviewed changes

cloud-fan reviewed Dec 28, 2017

View reviewed changes

Add override for stageId, stageAttemptId and partitionId.

72a3abf

Updating wording in TaskContext.stageAttemptId

cloud-fan reviewed Dec 29, 2017

View reviewed changes

Rename stageAttemptId to stageAttemptNumber

9266cd8

advancedxy force-pushed the SPARK-22897 branch from 2581263 to 9266cd8 Compare December 29, 2017 15:12

cloud-fan reviewed Jan 2, 2018

View reviewed changes

asfgit closed this in a6fc300 Jan 2, 2018

advancedxy deleted the SPARK-22897 branch January 2, 2018 15:33

tgravescs mentioned this pull request Jun 22, 2018

[SPARK-22897][CORE] Expose stageAttemptId in TaskContext #21609

Closed

turboFei mentioned this pull request Aug 31, 2018

[SPARK-22897][CORE] Expose stageAttemptId in TaskContext turboFei/spark#1

Closed

[SPARK-22897][CORE]: Expose stageAttemptId in TaskContext #20082

[SPARK-22897][CORE]: Expose stageAttemptId in TaskContext #20082

Uh oh!

Conversation

advancedxy commented Dec 26, 2017

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

advancedxy commented Dec 26, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jiangxb1987 commented Dec 26, 2017

Uh oh!

cloud-fan commented Dec 26, 2017

Uh oh!

SparkQA commented Dec 26, 2017

Uh oh!

advancedxy commented Dec 26, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 26, 2017

Uh oh!

SparkQA commented Dec 27, 2017

Uh oh!

SparkQA commented Dec 27, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 27, 2017

Uh oh!

zsxwing left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 28, 2017

Uh oh!

jiangxb1987 commented Dec 28, 2017

Uh oh!

SparkQA commented Dec 28, 2017

Uh oh!

advancedxy commented Dec 29, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 29, 2017

Uh oh!

SparkQA commented Dec 29, 2017

Uh oh!

advancedxy commented Jan 2, 2018

Uh oh!

Choose a reason for hiding this comment