Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13708#discussion_r69609019
--- Diff: core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala ---
@@ -17,61 +17,323 @@
package org.apache.spark.ui.jobs
+import java.net.URLEncoder
import java.util.Date
+import javax.servlet.http.HttpServletRequest
-import scala.xml.{Node, Text}
+import scala.collection.JavaConverters._
+import scala.xml._
import org.apache.commons.lang3.StringEscapeUtils
import org.apache.spark.scheduler.StageInfo
-import org.apache.spark.ui.{ToolTips, UIUtils}
+import org.apache.spark.ui._
+import org.apache.spark.ui.jobs.UIData.StageUIData
import org.apache.spark.util.Utils
-/** Page showing list of all ongoing and recently finished stages */
private[ui] class StageTableBase(
+ request: HttpServletRequest,
+ stages: Seq[StageInfo],
+ stageTag: String,
+ basePath: String,
+ subPath: String,
+ progressListener: JobProgressListener,
+ isFairScheduler: Boolean,
+ killEnabled: Boolean,
+ isFailedStage: Boolean) {
+ val allParameters = request.getParameterMap().asScala.toMap
+ val parameterOtherTable =
allParameters.filterNot(_._1.startsWith(stageTag))
+ .map(para => para._1 + "=" + para._2(0))
+
+ val parameterStagePage = request.getParameter(stageTag + ".page")
+ val parameterStageSortColumn = request.getParameter(stageTag + ".sort")
+ val parameterStageSortDesc = request.getParameter(stageTag + ".desc")
+ val parameterStagePageSize = request.getParameter(stageTag + ".pageSize")
+ val parameterStagePrevPageSize = request.getParameter(stageTag +
".prevPageSize")
+
+ val stagePage = Option(parameterStagePage).map(_.toInt).getOrElse(1)
+ val stageSortColumn = Option(parameterStageSortColumn).map { sortColumn
=>
+ UIUtils.decodeURLParameter(sortColumn)
+ }.getOrElse("Stage Id")
+ val stageSortDesc =
Option(parameterStageSortDesc).map(_.toBoolean).getOrElse(
+ // New stages should be shown above old jobs by default.
+ if (stageSortColumn == "Stage Id") true else false
+ )
+ val stagePageSize =
Option(parameterStagePageSize).map(_.toInt).getOrElse(100)
+ val stagePrevPageSize = Option(parameterStagePrevPageSize).map(_.toInt)
+ .getOrElse(stagePageSize)
+
+ val page: Int = {
+ // If the user has changed to a larger page size, then go to page 1 in
order to avoid
+ // IndexOutOfBoundsException.
+ if (stagePageSize <= stagePrevPageSize) {
+ stagePage
+ } else {
+ 1
+ }
+ }
+ val currentTime = System.currentTimeMillis()
+
+ val toNodeSeq = try {
+ new StagePagedTable(
+ stages,
+ stageTag,
+ basePath,
+ subPath,
+ progressListener,
+ isFairScheduler,
+ killEnabled,
+ currentTime,
+ stagePageSize,
+ stageSortColumn,
+ stageSortDesc,
+ isFailedStage,
+ parameterOtherTable
+ ).table(page)
+ } catch {
+ case e @ (_ : IllegalArgumentException | _ :
IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering stage table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+}
+
+private[ui] class StageTableRowData(
+ val stageInfo: StageInfo,
+ val stageData: Option[StageUIData],
+ val stageId: Int,
+ val attemptId: Int,
+ val schedulingPool: String,
+ val description: String,
+ val descriptionOption: Option[String],
+ val submissionTime: Long,
+ val formattedSubmissionTime: String,
+ val duration: Long,
+ val formattedDuration: String,
+ val inputRead: Long,
+ val inputReadWithUnit: String,
+ val outputWrite: Long,
+ val outputWriteWithUnit: String,
+ val shuffleRead: Long,
+ val shuffleReadWithUnit: String,
+ val shuffleWrite: Long,
+ val shuffleWriteWithUnit: String)
+
+private[ui] class MissingStageTableRowData(
+ stageInfo: StageInfo,
+ stageId: Int,
+ attemptId: Int) extends StageTableRowData(
+ stageInfo, None, stageId, attemptId, "", "", None, 0, "", -1, "", 0, "",
0, "", 0, "", 0, "")
+
+/** Page showing list of all ongoing and recently finished stages */
+private[ui] class StagePagedTable(
stages: Seq[StageInfo],
+ stageTag: String,
basePath: String,
+ subPath: String,
listener: JobProgressListener,
isFairScheduler: Boolean,
- killEnabled: Boolean) {
-
- protected def columns: Seq[Node] = {
- <th>Stage Id</th> ++
- {if (isFairScheduler) {<th>Pool Name</th>} else Seq.empty} ++
- <th>Description</th>
- <th>Submitted</th>
- <th>Duration</th>
- <th>Tasks: Succeeded/Total</th>
- <th><span data-toggle="tooltip"
title={ToolTips.INPUT}>Input</span></th>
- <th><span data-toggle="tooltip"
title={ToolTips.OUTPUT}>Output</span></th>
- <th><span data-toggle="tooltip" title={ToolTips.SHUFFLE_READ}>Shuffle
Read</span></th>
- <th>
- <!-- Place the shuffle write tooltip on the left (rather than the
default position
- of on top) because the shuffle write column is the last column on
the right side and
- the tooltip is wider than the column, so it doesn't fit on top. -->
- <span data-toggle="tooltip" data-placement="left"
title={ToolTips.SHUFFLE_WRITE}>
- Shuffle Write
- </span>
- </th>
+ killEnabled: Boolean,
+ currentTime: Long,
+ pageSize: Int,
+ sortColumn: String,
+ desc: Boolean,
+ isFailedStage: Boolean,
+ parameterOtherTable: Iterable[String]) extends
PagedTable[StageTableRowData] {
+
+ override def tableId: String = stageTag + "-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-condensed table-striped
table-head-clickable"
+
+ override def pageSizeFormField: String = stageTag + ".pageSize"
+
+ override def prevPageSizeFormField: String = stageTag + ".prevPageSize"
+
+ override def pageNumberFormField: String = stageTag + ".page"
+
+ val parameterPath = UIUtils.prependBaseUri(basePath) + s"/$subPath/?" +
+ parameterOtherTable.mkString("&")
+
+ override val dataSource = new StageDataSource(
+ stages,
+ listener,
+ currentTime,
+ pageSize,
+ sortColumn,
+ desc
+ )
+
+ override def pageLink(page: Int): String = {
+ val encodedSortColumn = URLEncoder.encode(sortColumn, "UTF-8")
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$stageTag.sort=$encodedSortColumn" +
+ s"&$stageTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize"
+ }
+
+ override def goButtonFormPath: String = {
+ val encodedSortColumn = URLEncoder.encode(sortColumn, "UTF-8")
+
s"$parameterPath&$stageTag.sort=$encodedSortColumn&$stageTag.desc=$desc"
}
- def toNodeSeq: Seq[Node] = {
- listener.synchronized {
- stageTable(renderStageRow, stages)
+ override def headers: Seq[Node] = {
+ val stageHeadersAndCssClasses: Seq[(String, Option[(String, Boolean)],
Boolean)] =
--- End diff --
nit: could you add a comment to explain the meaning of `(String,
Option[(String, Boolean)], Boolean)`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
