[SPARK-20464][SS] Add a job group and description for streaming queries and fix cancellation of running jobs using the job group #17765

kunalkhamar · 2017-04-25T22:26:51Z

What changes were proposed in this pull request?

Job group: adding a job group is required to properly cancel running jobs related to a query.
Description: the new description makes it easier to group the batches of a query by sorting by name in the Spark Jobs UI.

How was this patch tested?

Unit tests
UI screenshot
- Order by job id:
- Order by description:
- Order by job id (no query name):
- Order by description (no query name):

SparkQA · 2017-04-26T00:49:43Z

Test build #76152 has finished for PR 17765 at commit 07e182b.

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

SparkQA · 2017-04-26T02:56:13Z

Test build #76160 has finished for PR 17765 at commit bd13a01.

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

SparkQA · 2017-04-26T02:57:28Z

Test build #76162 has finished for PR 17765 at commit 609d50e.

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

SparkQA · 2017-04-26T03:07:44Z

Test build #76163 has finished for PR 17765 at commit bd13a01.

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

zsxwing · 2017-04-26T20:42:40Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

+  private def getBatchDescriptionString: String = {
+    val batchDescription = if (currentBatchId < 0) "init" else currentBatchId.toString
+    Option(name).map(_ + " ").getOrElse("") +
+      s"[batch = $batchDescription, id = $id, runId = $runId]"


The description looks good to me. @marmbrus what do you think?

marmbrus · 2017-04-26T20:54:18Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

   */
  private def runBatches(): Unit = {
    try {
+      sparkSession.sparkContext.setJobGroup(runId.toString, getBatchDescriptionString)


@brkyvz is this okay?

yes, this seems fine

note here it is setJobGroup(..., interruptOnCancel = false). Should we add a sql conf for interruptOnCancel?

when would you want to set this to true or false?

Had this set to false due to HDFS-1208, but setting it to true since the HDFS bug is 7 years old.

SparkQA · 2017-04-26T22:32:04Z

Test build #76198 has finished for PR 17765 at commit 7a58547.

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

…ts to be processed.

kunalkhamar · 2017-04-27T23:36:37Z

core/src/main/scala/org/apache/spark/ui/UIUtils.scala


      // Verify that this has only anchors and span (we are wrapping in span)
-      val allowedNodeLabels = Set("a", "span")
+      val allowedNodeLabels = Set("a", "span", "br")


@tdas Is this change okay? Need it to add line breaks in the job description cells.

kunalkhamar · 2017-04-27T23:37:39Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

              if (currentBatchId < 0) {
                // We'll do this initialization only once
                populateStartOffsets(sparkSessionToRunBatches)
+                sparkSession.sparkContext.setJobDescription(getBatchDescriptionString)


Update job description with correct currentBatchId after initializing starting offsets.

kunalkhamar · 2017-04-27T23:38:07Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

              logDebug(s"batch ${currentBatchId} committed")
              // We'll increase currentBatchId after we complete processing current batch's data
              currentBatchId += 1
+              sparkSession.sparkContext.setJobDescription(getBatchDescriptionString)


Update job description with updated currentBatchId after each batch.

kunalkhamar · 2017-04-27T23:38:45Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

+  private def getBatchDescriptionString: String = {
+    val batchDescription = if (currentBatchId < 0) "init" else currentBatchId.toString
+    Option(name).map(_ + " ").getOrElse("") +
+      s"[batch = $batchDescription,<br/>id = $id,<br/>runId = $runId]"


@marmbrus @zsxwing @tdas Is this format good for description?

I would get rid of the [] if you are going to use newlines

Yes, updated.

kunalkhamar · 2017-04-28T00:26:41Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

+  private def getBatchDescriptionString: String = {
+    val batchDescription = if (currentBatchId < 0) "init" else currentBatchId.toString
+    Option(name).map(_ + "<br/>").getOrElse("") +
+      s"id = $id<br/>runId = $runId<br/>batch = $batchDescription"


@marmbrus @zsxwing @tdas Updated as per comments, the screenshots are in the PR description.

SparkQA · 2017-04-28T00:36:25Z

Test build #76246 has finished for PR 17765 at commit 992d68f.

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

SparkQA · 2017-04-28T00:36:50Z

Test build #76245 has finished for PR 17765 at commit 6ab66e2.

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

SparkQA · 2017-04-28T01:45:58Z

Test build #76249 has finished for PR 17765 at commit 915d67b.

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

SparkQA · 2017-04-28T02:31:09Z

Test build #76250 has finished for PR 17765 at commit f9342c9.

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

SparkQA · 2017-04-28T03:04:53Z

Test build #76251 has finished for PR 17765 at commit 6e66638.

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

brkyvz · 2017-04-28T20:45:49Z

LGTM! Any other comments @zsxwing?

zsxwing · 2017-05-01T18:25:31Z

LGTM!

zsxwing · 2017-05-01T18:37:10Z

Merging to master and 2.2.

…es and fix cancellation of running jobs using the job group ## What changes were proposed in this pull request? Job group: adding a job group is required to properly cancel running jobs related to a query. Description: the new description makes it easier to group the batches of a query by sorting by name in the Spark Jobs UI. ## How was this patch tested? - Unit tests - UI screenshot - Order by job id: ![screen shot 2017-04-27 at 5 10 09 pm](https://cloud.githubusercontent.com/assets/7865120/25509468/15452274-2b6e-11e7-87ba-d929816688cf.png) - Order by description: ![screen shot 2017-04-27 at 5 10 22 pm](https://cloud.githubusercontent.com/assets/7865120/25509474/1c298512-2b6e-11e7-99b8-fef1ef7665c1.png) - Order by job id (no query name): ![screen shot 2017-04-27 at 5 21 33 pm](https://cloud.githubusercontent.com/assets/7865120/25509482/28c96dc8-2b6e-11e7-8df0-9d3cdbb05e36.png) - Order by description (no query name): ![screen shot 2017-04-27 at 5 21 44 pm](https://cloud.githubusercontent.com/assets/7865120/25509489/37674742-2b6e-11e7-9357-b5c38ec16ac4.png) Author: Kunal Khamar <[email protected]> Closes #17765 from kunalkhamar/sc-6696. (cherry picked from commit 6fc6cf8) Signed-off-by: Shixiong Zhu <[email protected]>

…es and fix cancellation of running jobs using the job group ## What changes were proposed in this pull request? Job group: adding a job group is required to properly cancel running jobs related to a query. Description: the new description makes it easier to group the batches of a query by sorting by name in the Spark Jobs UI. ## How was this patch tested? - Unit tests - UI screenshot - Order by job id: ![screen shot 2017-04-27 at 5 10 09 pm](https://cloud.githubusercontent.com/assets/7865120/25509468/15452274-2b6e-11e7-87ba-d929816688cf.png) - Order by description: ![screen shot 2017-04-27 at 5 10 22 pm](https://cloud.githubusercontent.com/assets/7865120/25509474/1c298512-2b6e-11e7-99b8-fef1ef7665c1.png) - Order by job id (no query name): ![screen shot 2017-04-27 at 5 21 33 pm](https://cloud.githubusercontent.com/assets/7865120/25509482/28c96dc8-2b6e-11e7-8df0-9d3cdbb05e36.png) - Order by description (no query name): ![screen shot 2017-04-27 at 5 21 44 pm](https://cloud.githubusercontent.com/assets/7865120/25509489/37674742-2b6e-11e7-9357-b5c38ec16ac4.png) Author: Kunal Khamar <[email protected]> Closes apache#17765 from kunalkhamar/sc-6696. (cherry picked from commit 6fc6cf8)

kunalkhamar added 2 commits April 25, 2017 15:23

Set job group for streaming query batches.

07e182b

Add cancel by job group.

bd13a01

kunalkhamar force-pushed the sc-6696 branch from 609d50e to bd13a01 Compare April 26, 2017 00:48

kunalkhamar changed the title ~~[SPARK-20464][SS] Add a job group and an informative description for streaming queries~~ [SPARK-20464][SS] Add a job group and description for streaming queries and cancel running jobs using the job group Apr 26, 2017

kunalkhamar changed the title ~~[SPARK-20464][SS] Add a job group and description for streaming queries and cancel running jobs using the job group~~ [SPARK-20464][SS] Add a job group and description for streaming queries and fix cancellation of running jobs using the job group Apr 26, 2017

Add test for job cancellation.

7a58547

zsxwing reviewed Apr 26, 2017

View reviewed changes

marmbrus reviewed Apr 26, 2017

View reviewed changes

kunalkhamar added 3 commits April 27, 2017 15:25

Add test for job cancellation.

6ab66e2

Remove extra setJobDesc.

992d68f

Update UI with line breaks, update test to wait for listener bus even…

915d67b

…ts to be processed.

kunalkhamar commented Apr 27, 2017

View reviewed changes

kunalkhamar added 2 commits April 27, 2017 16:41

Stop query in test.

f9342c9

Update batch desc format.

6e66638

kunalkhamar commented Apr 28, 2017

View reviewed changes

asfgit closed this in 6fc6cf8 May 1, 2017

[SPARK-20464][SS] Add a job group and description for streaming queries and fix cancellation of running jobs using the job group #17765

[SPARK-20464][SS] Add a job group and description for streaming queries and fix cancellation of running jobs using the job group #17765

Uh oh!

Conversation

kunalkhamar commented Apr 25, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

SparkQA commented Apr 26, 2017

Uh oh!

SparkQA commented Apr 26, 2017

Uh oh!

SparkQA commented Apr 26, 2017

Uh oh!

SparkQA commented Apr 26, 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!

zsxwing Apr 27, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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 Apr 26, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kunalkhamar Apr 27, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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 Apr 28, 2017

Uh oh!

SparkQA commented Apr 28, 2017

Uh oh!

SparkQA commented Apr 28, 2017

Uh oh!

SparkQA commented Apr 28, 2017

Uh oh!

SparkQA commented Apr 28, 2017

Uh oh!

brkyvz commented Apr 28, 2017

Uh oh!

zsxwing commented May 1, 2017

Uh oh!

zsxwing commented May 1, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

kunalkhamar commented Apr 25, 2017 •

edited

Loading

zsxwing Apr 27, 2017 •

edited

Loading

kunalkhamar Apr 27, 2017 •

edited

Loading