Skip to content

[SPARK-21922] Fix duration always updating when task failed but status is still RUN… #19132

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
caneGuy wants to merge 8 commits into from
Closed

[SPARK-21922] Fix duration always updating when task failed but status is still RUN… #19132

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
03a33a9
Fix duration always updating when task failed but status is still RUN…
caneGuy Sep 5, 2017
a442000
Revert "Fix duration always updating when task failed but status is s…
caneGuy Sep 6, 2017
8b47418
Do not access fs in SPARK-UI pass event log access time from FSHistor…
caneGuy Sep 6, 2017
25fe22c
Update as commented
caneGuy Sep 12, 2017
560d442
Update as commented
caneGuy Sep 13, 2017
f0e1ded
Make api more cleaner
caneGuy Sep 13, 2017
8ec4319
Modify field name and update code style
caneGuy Sep 14, 2017
06035f7
Update code style
caneGuy Sep 14, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -249,7 +249,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
val appSecManager = new SecurityManager(conf)
SparkUI.createHistoryUI(conf, replayBus, appSecManager, appInfo.name,
HistoryServer.getAttemptURI(appId, attempt.attemptId),
attempt.startTime)
attempt.lastUpdated, attempt.startTime)
// Do not call ui.bind() to avoid creating a new server for each application
}

Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/org/apache/spark/status/api/v1/AllStagesResource.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,7 +142,7 @@ private[v1] object AllStagesResource {
index = uiData.taskInfo.index,
attempt = uiData.taskInfo.attemptNumber,
launchTime = new Date(uiData.taskInfo.launchTime),
duration = uiData.taskDuration,
duration = uiData.taskDuration(),

@ajbozarth ajbozarth Sep 6, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is unrelated and unnecessary

@caneGuy caneGuy Sep 6, 2017 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since i have changed taskDuration below,if do not add () here a compiler error will occur.

@ajbozarth ajbozarth Sep 6, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

odd, I thought for sure it'd be fine, then this LGTM

@caneGuy caneGuy Sep 6, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @ajbozarth

@jerryshao jerryshao Sep 13, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here what if we call the REST API on history server to get stage info? Looks like we may still have this issue since we don't have last update time here, what do you think @ajbozarth ?

@caneGuy caneGuy Sep 13, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are right, @jerryshao .IIUC, the ui in AllStagesResource.scala is passed from ApiRootResource which also create sparkUI by FSHistoryProvider.So we can also get lastUpdateTime from this ui in AllStagesResource and pass to the taskDuration interface.I think it is another problem for REST?Should we fix here?

@jerryshao jerryshao Sep 13, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, if it is not a big change I think it should be fixed here. Because currently with this fix UI and REST API are inconsistent.

executorId = uiData.taskInfo.executorId,
host = uiData.taskInfo.host,
status = uiData.taskInfo.status,
Expand Down
8 changes: 6 additions & 2 deletions core/src/main/scala/org/apache/spark/ui/SparkUI.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@ private[spark] class SparkUI private (
val operationGraphListener: RDDOperationGraphListener,
var appName: String,
val basePath: String,
val lastUpdateTime: Long = -1L,

@jerryshao jerryshao Sep 12, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would like to user Option[Long] = None as default value to reflect there's no update time.

@caneGuy caneGuy Sep 12, 2017 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated @jerryshao Thanks for your time.

val startTime: Long)
extends WebUI(securityManager, securityManager.getSSLOptions("ui"), SparkUI.getUIPort(conf),
conf, basePath, "SparkUI")
Expand Down Expand Up @@ -176,9 +177,11 @@ private[spark] object SparkUI {
securityManager: SecurityManager,
appName: String,
basePath: String,
lastUpdateTime: Long,
startTime: Long): SparkUI = {
val sparkUI = create(
None, conf, listenerBus, securityManager, appName, basePath, startTime = startTime)
None, conf, listenerBus, securityManager, appName, basePath,
lastUpdateTime = lastUpdateTime, startTime = startTime)

val listenerFactories = ServiceLoader.load(classOf[SparkHistoryListenerFactory],
Utils.getContextOrSparkClassLoader).asScala
Expand All @@ -204,6 +207,7 @@ private[spark] object SparkUI {
appName: String,
basePath: String = "",
jobProgressListener: Option[JobProgressListener] = None,
lastUpdateTime: Long = -1L,
startTime: Long): SparkUI = {

val _jobProgressListener: JobProgressListener = jobProgressListener.getOrElse {
Expand All @@ -226,6 +230,6 @@ private[spark] object SparkUI {

new SparkUI(sc, conf, securityManager, environmentListener, storageStatusListener,
executorsListener, _jobProgressListener, storageListener, operationGraphListener,
appName, basePath, startTime)
appName, basePath, lastUpdateTime, startTime)
}
}
9 changes: 7 additions & 2 deletions core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -299,6 +299,7 @@ private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {
stageData.hasShuffleRead,
stageData.hasShuffleWrite,
stageData.hasBytesSpilled,
parent.lastUpdateTime,
currentTime,
pageSize = taskPageSize,
sortColumn = taskSortColumn,
Expand Down Expand Up @@ -863,6 +864,7 @@ private[ui] class TaskDataSource(
hasShuffleRead: Boolean,
hasShuffleWrite: Boolean,
hasBytesSpilled: Boolean,
lastUpdateTime: Long,
currentTime: Long,
pageSize: Int,
sortColumn: String,
Expand All @@ -889,8 +891,9 @@ private[ui] class TaskDataSource(
private def taskRow(taskData: TaskUIData): TaskTableRowData = {
val info = taskData.taskInfo
val metrics = taskData.metrics
val duration = taskData.taskDuration.getOrElse(1L)
val formatDuration = taskData.taskDuration.map(d => UIUtils.formatDuration(d)).getOrElse("")
val duration = taskData.taskDuration(lastUpdateTime).getOrElse(1L)
val formatDuration =
taskData.taskDuration(lastUpdateTime).map(d => UIUtils.formatDuration(d)).getOrElse("")
val schedulerDelay = metrics.map(getSchedulerDelay(info, _, currentTime)).getOrElse(0L)
val gcTime = metrics.map(_.jvmGCTime).getOrElse(0L)
val taskDeserializationTime = metrics.map(_.executorDeserializeTime).getOrElse(0L)
Expand Down Expand Up @@ -1154,6 +1157,7 @@ private[ui] class TaskPagedTable(
hasShuffleRead: Boolean,
hasShuffleWrite: Boolean,
hasBytesSpilled: Boolean,
lastUpdateTime: Long,
currentTime: Long,
pageSize: Int,
sortColumn: String,
Expand All @@ -1179,6 +1183,7 @@ private[ui] class TaskPagedTable(
hasShuffleRead,
hasShuffleWrite,
hasBytesSpilled,
lastUpdateTime,
currentTime,
pageSize,
sortColumn,
Expand Down
1 change: 1 addition & 0 deletions core/src/main/scala/org/apache/spark/ui/jobs/StagesTab.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ private[ui] class StagesTab(parent: SparkUI) extends SparkUITab(parent, "stages"
val progressListener = parent.jobProgressListener
val operationGraphListener = parent.operationGraphListener
val executorsListener = parent.executorsListener
val lastUpdateTime = parent.lastUpdateTime

attachPage(new AllStagesPage(this))
attachPage(new StagePage(this))
Expand Down
5 changes: 3 additions & 2 deletions core/src/main/scala/org/apache/spark/ui/jobs/UIData.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,9 +133,10 @@ private[spark] object UIData {
_metrics = metrics.map(TaskMetricsUIData.fromTaskMetrics)
}

def taskDuration: Option[Long] = {
def taskDuration(lastUpdateTime: Long = -1L): Option[Long] = {
if (taskInfo.status == "RUNNING") {
Some(_taskInfo.timeRunning(System.currentTimeMillis))
Some(_taskInfo.timeRunning(
if (lastUpdateTime == -1) System.currentTimeMillis else lastUpdateTime))
} else {
_metrics.map(_.executorRunTime)
}
Expand Down