Spark Core源码精读计划14 | Spark Web UI界面的实现

推荐阅读

《Spark源码精度计划1 | SparkConf》

《Spark Core源码精读计划2 | SparkContext组件初始化》

《Spark Core源码精读计划3 | SparkContext辅助属性及后初始化》

《Spark Core源码精读计划4 | SparkContext提供的其他功能》

《Spark Core源码精读计划5 | 事件总线及ListenerBus》

《Spark Core源码精读计划6 | AsyncEventQueue与LiveListenerBus》

《Spark Core源码精读计划7 | Spark执行环境的初始化》

《Spark Core源码精读计划8 | SparkEnv中RPC环境的基础构建》

《Spark Core源码精读计划9 | Spark RPC环境中的消息调度逻辑》

《Spark Core源码精读计划10 | NettyRpcEnv客户端消息发送逻辑》

《Spark Core源码精读计划11 | Spark广播机制的实现》

《Spark Core源码精读计划12 | Spark序列化及压缩机制浅析》

《Spark Core源码精读计划13 | 度量系统MetricsSystem的建立》

目录

  • 前言
  • 创建SparkUI
    • SparkContext中的操作
    • 初始化SparkUI
  • WebUI的具体实现
    • 属性成员和Getter方法
    • WebUI提供的attach/detach类方法
    • 绑定WebUI到Jetty服务
  • Spark Web UI的展示
    • WebUITab与WebUIPage的定义
    • 渲染Spark UI页面
  • 总结

前言

我们已经在SparkEnv的世界里摸爬滚打了很长时间,对RPC环境、广播变量、序列化和压缩、度量系统这几个相对独立的组件有了一定的了解。现在是时候抽身出来,继续move forward,跟着SparkContext初始化的流程走下去。按照顺序,本文要讲的是Spark Web UI。正好,上一篇文章刚刚讲过度量系统,本文可以说是水到渠成了。

Spark Web UI主要依赖于流行的Servlet容器Jetty实现,本文为避免跑题,在涉及Jetty相关细节的时候都不会详细地展开。

创建SparkUI

由于距离讲SparkContext的初始化已经过去许久了,因此先看看SparkUI在SparkContext的创建流程。

SparkContext中的操作

代码#14.1 - SparkContext中创建SparkUI的代码

    _ui =
      if (conf.getBoolean("spark.ui.enabled", true)) {
        Some(SparkUI.create(Some(this), _statusStore, _conf, _env.securityManager, appName, "",
          startTime))
      } else {
        None
      }
    _ui.foreach(_.bind())

其中,_statusStore是早先初始化的AppStatusStore,它是包装过的KVStore和AppStatusListener,前者用于存储监控数据,后者注册到事件总线中的appStatus队列中。_env.securityManager则是SparkEnv中初始化的安全管理器。

SparkContext通过调用SparkUI伴生对象中的create()方法来直接new出SparkUI实例,然后调用bind()方法将SparkUI绑定到Jetty服务。bind()方法之后再说,现在先来看SparkUI类的事情。

初始化SparkUI

以下是SparkUI类中的属性成员,以及构造方法。

代码#14.2 - o.a.s.ui.SparkUI类中的成员属性和initialize()方法

private[spark] class SparkUI private (
    val store: AppStatusStore,
    val sc: Option[SparkContext],
    val conf: SparkConf,
    securityManager: SecurityManager,
    var appName: String,
    val basePath: String,
    val startTime: Long,
    val appSparkVersion: String)
  extends WebUI(securityManager, securityManager.getSSLOptions("ui"), SparkUI.getUIPort(conf),
    conf, basePath, "SparkUI")
  with Logging with UIRoot {
  val killEnabled = sc.map(_.conf.getBoolean("spark.ui.killEnabled", true)).getOrElse(false)
  var appId: String = _
  private var streamingJobProgressListener: Option[SparkListener] = None

  def initialize(): Unit = {
    val jobsTab = new JobsTab(this, store)
    attachTab(jobsTab)
    val stagesTab = new StagesTab(this, store)
    attachTab(stagesTab)
    attachTab(new StorageTab(this, store))
    attachTab(new EnvironmentTab(this, store))
    attachTab(new ExecutorsTab(this))
    attachHandler(createStaticHandler(SparkUI.STATIC_RESOURCE_DIR, "/static"))
    attachHandler(createRedirectHandler("/", "/jobs/", basePath = basePath))
    attachHandler(ApiRootResource.getServletHandler(this))

    attachHandler(createRedirectHandler(
      "/jobs/job/kill", "/jobs/", jobsTab.handleKillRequest, httpMethods = Set("GET", "POST")))
    attachHandler(createRedirectHandler(
      "/stages/stage/kill", "/stages/", stagesTab.handleKillRequest,
      httpMethods = Set("GET", "POST")))
  }

  initialize()
}

SparkUI类中有3个属性成员:

  • killEnabled由配置项spark.ui.killEnabled控制,如果为true,会在UI界面中展示强行杀掉Spark Job的开关。
  • appId就是当前的Application ID。
  • streamingJobProgressListener是用于Spark Streaming作业进度的监听器。

在initialize()方法中,首先创建了5个Tab,并调用了attachTab()方法注册到Web UI。所谓Tab就是Spark UI中的标签页,如下图中最上面的一栏所示,名称也是一一对应的。

图#14.1 - Spark Web UI页面

接下来,调用createStaticHandler()方法创建静态资源的ServletContextHandler,又调用createRedirectHandler()创建一些重定向的ServletContextHandler。【插一句:ServletContextHandler是Jetty中一个功能完善的处理器,负责接收并处理HTTP请求,再投递给Servlet。】最后,逐一调用attachHandler()方法注册到Web UI。

那么上面的这一系列方法(也包含上一节的bind()方法)是哪儿来的呢?答案是WebUI抽象类,也就是SparkUI的基类。下面来阅读它的源码。

WebUI的具体实现

WebUI是Spark里所有可以在浏览器中展示的内容的顶级组件,因此SparkUI类也会继承它。

属性成员和Getter方法

代码#14.3 - o.a.s.ui.WebUI类的属性成员和Getter方法

protected val tabs = ArrayBuffer[WebUITab]()
  protected val handlers = ArrayBuffer[ServletContextHandler]()
  protected val pageToHandlers = new HashMap[WebUIPage, ArrayBuffer[ServletContextHandler]]
  protected var serverInfo: Option[ServerInfo] = None
  protected val publicHostName = Option(conf.getenv("SPARK_PUBLIC_DNS")).getOrElse(
    conf.get(DRIVER_HOST_ADDRESS))
  private val className = Utils.getFormattedClassName(this)

  def getBasePath: String = basePath
  def getTabs: Seq[WebUITab] = tabs
  def getHandlers: Seq[ServletContextHandler] = handlers
  def getSecurityManager: SecurityManager = securityManager

属性成员有以下6个。

  • tabs:持有WebUITab(即图#14.1中的标签页)的缓存。
  • handlers:持有Jetty ServletContextHandler的缓存。
  • pageToHandlers:保存WebUIPage(WebUITab的下一级组件)与其对应的ServletContextHandler的映射关系。
  • serverInfo:当前Web UI对应的Jetty服务器信息。
  • publicHostName:当前Web UI对应的Jetty服务主机名。先通过系统环境变量SPARK_PUBLIC_DNS获取,再通过spark.driver.host配置项获取。
  • className:当前类的名称,用Utils.getFormattedClassName()方法格式化过。

Getter方法有4个,getTabs()和getHandlers()都是简单地获得对应属性的值。getBasePath()取得构造参数中定义的Web UI基路径,getSecurityManager()则取得构造参数中传入的安全管理器。

WebUI提供的attach/detach类方法

这类方法都是成对的,一共有3对:attachTab()/detachTab(),用于注册和移除WebUITab;attachPage()/detachPage(),用于注册和移除WebUIPage;attachHandler()/detachHandler(),用于注册和移除ServletContextHandler。以下是它们的代码。

代码#14.4 - WebUI提供的attach/detach类方法

  def attachTab(tab: WebUITab) {
    tab.pages.foreach(attachPage)
    tabs += tab
  }

  def detachTab(tab: WebUITab) {
    tab.pages.foreach(detachPage)
    tabs -= tab
  }

  def detachPage(page: WebUIPage) {
    pageToHandlers.remove(page).foreach(_.foreach(detachHandler))
  }

  def attachPage(page: WebUIPage) {
    val pagePath = "/" + page.prefix
    val renderHandler = createServletHandler(pagePath,
      (request: HttpServletRequest) => page.render(request), securityManager, conf, basePath)
    val renderJsonHandler = createServletHandler(pagePath.stripSuffix("/") + "/json",
      (request: HttpServletRequest) => page.renderJson(request), securityManager, conf, basePath)
    attachHandler(renderHandler)
    attachHandler(renderJsonHandler)
    val handlers = pageToHandlers.getOrElseUpdate(page, ArrayBuffer[ServletContextHandler]())
    handlers += renderHandler
  }

  def attachHandler(handler: ServletContextHandler) {
    handlers += handler
    serverInfo.foreach(_.addHandler(handler))
  }

  def detachHandler(handler: ServletContextHandler) {
    handlers -= handler
    serverInfo.foreach(_.removeHandler(handler))
  }

看起来并不难理解,我们就来读读其中最长的attachPage()方法。它的流程是:调用Jetty工具类JettyUtils的createServletHander()方法,为WebUIPage的两个渲染方法render()和renderJson()创建ServletContextHandler,也就是一个WebUIPage需要对应两个处理器。然后,调用上述attachHandler()方法向Jetty注册处理器,并将映射关系写入handlers结构中。绑定WebUI到Jetty服务

这里就是在前一章节提到的bind()方法了。

代码#14.5 - o.a.s.ui.WebUI.bind()方法

  def bind(): Unit = {
    assert(serverInfo.isEmpty, s"Attempted to bind $className more than once!")
    try {
      val host = Option(conf.getenv("SPARK_LOCAL_IP")).getOrElse("0.0.0.0")
      serverInfo = Some(startJettyServer(host, port, sslOptions, handlers, conf, name))
      logInfo(s"Bound $className to $host, and started at $webUrl")
    } catch {
      case e: Exception =>
        logError(s"Failed to bind $className", e)
        System.exit(1)
    }
  }

该方法调用了JettyUtils.startJettyServer()方法来启动Jetty服务,具体不再赘述。

Spark Web UI的展示

Spark Web UI实际上是一个三层的树形结构,根节点为WebUI,中层节点为WebUITab,叶子节点为WebUIPage。UI界面的展示就主要靠WebUITab与WebUIPage来实现。在Spark UI界面中,一个Tab可以包含一个或多个Page,并且Tab是可选的。

WebUITab与WebUIPage的定义

以下是WebUITab的代码。

代码#14.6 - o.a.s.ui.WebUITab抽象类

private[spark] abstract class WebUITab(parent: WebUI, val prefix: String) {
  val pages = ArrayBuffer[WebUIPage]()
  val name = prefix.capitalize

  def attachPage(page: WebUIPage) {
    page.prefix = (prefix + "/" + page.prefix).stripSuffix("/")
    pages += page
  }

  def headerTabs: Seq[WebUITab] = parent.getTabs

  def basePath: String = parent.getBasePath
}

由于一个Tab可以包含多个Page,因此pages数组就用来缓存该Tab下所有的Page。attachPage()方法就用于将Tab的路径前缀与Page的路径前缀拼合起来,并将其加入pages数组中。

WebUIPage抽象类的定义更加简单,只有两个方法,前面已经出现过。render()方法用于渲染页面,renderJson()方法则用于生成对应的JSON串,代码就不再贴出来了。

WebUITab与WebUIPage各有很多的实现类,分别对应一个Tab或一个Page。本来想拿IDEA生成两张类图,但是不知为何,所有表示继承关系的箭头都显示不出来(可能IDEA对Scala的支持仍然不是很好吧),只得作罢。最后,我们来看看Spark UI上的内容是怎样展示出来的。

渲染Spark UI页面

我们以Environment这一页为例来探索,因为它的页面元素相当简单,只是展示许多环境信息(如Spark配置、系统属性、JVM信息、Classpath等等)的表格,干扰比较少。其页面本身如下图所示。

图#14.2 - Spark UI Environment页

首先来看EnvironmentTab的代码,非常简单。

代码#14.7 - o.a.s.ui.env.EnvironmentTab类

private[ui] class EnvironmentTab(
    parent: SparkUI,
    store: AppStatusStore) extends SparkUITab(parent, "environment") {
  attachPage(new EnvironmentPage(this, parent.conf, store))
}

其中SparkUITab就是对WebUITab的简单封装,加上了Application名称和Spark版本的属性。EnvironmentTab类只有构造方法,调用代码#14.6中预先定义好的attachPage()方法,将EnvironmentPage加入。以下则是EnvironmentPage的具体实现。

代码#14.8 - o.a.s.ui.env.EnvironmentPage类

private[ui] class EnvironmentPage(
    parent: EnvironmentTab,
    conf: SparkConf,
    store: AppStatusStore) extends WebUIPage("") {

  def render(request: HttpServletRequest): Seq[Node] = {
    val appEnv = store.environmentInfo()
    val jvmInformation = Map(
      "Java Version" -> appEnv.runtime.javaVersion,
      "Java Home" -> appEnv.runtime.javaHome,
      "Scala Version" -> appEnv.runtime.scalaVersion)
    val runtimeInformationTable = UIUtils.listingTable(
      propertyHeader, jvmRow, jvmInformation, fixedWidth = true)
    val sparkPropertiesTable = UIUtils.listingTable(propertyHeader, propertyRow,
      Utils.redact(conf, appEnv.sparkProperties.toSeq), fixedWidth = true)
    val systemPropertiesTable = UIUtils.listingTable(
      propertyHeader, propertyRow, appEnv.systemProperties, fixedWidth = true)
    val classpathEntriesTable = UIUtils.listingTable(
      classPathHeaders, classPathRow, appEnv.classpathEntries, fixedWidth = true)
    val content =
      <span>
        <h4>Runtime Information</h4> {runtimeInformationTable}
        <h4>Spark Properties</h4> {sparkPropertiesTable}
        <h4>System Properties</h4> {systemPropertiesTable}
        <h4>Classpath Entries</h4> {classpathEntriesTable}
      </span>
    UIUtils.headerSparkPage("Environment", content, parent)
  }

  private def propertyHeader = Seq("Name", "Value")
  private def classPathHeaders = Seq("Resource", "Source")
  private def jvmRow(kv: (String, String)) = <tr><td>{kv._1}</td><td>{kv._2}</td></tr>
  private def propertyRow(kv: (String, String)) = <tr><td>{kv._1}</td><td>{kv._2}</td></tr>
  private def classPathRow(data: (String, String)) = <tr><td>{data._1}</td><td>{data._2}</td></tr>
}

render()方法用来渲染页面内容,其流程如下:

  • 从AppStatusStore中取得所有环境信息。
  • 调用UIUtils.listingTable()方法,将对应的表头与添加了HTML标签的行封装成表格。
  • 将4张表格排列好,调用UIUtils.headerSparkPage()方法,按照规定好的页面布局展示在浏览器上。

这样,图#14.2的页面就显示出来了。

总结

本文从SparkContext中对Spark UI的初始化入手,首先介绍了SparkUI类的具体构造。然后分析了SparkUI的基类WebUI的具体实现,明确了整个UI界面的组成部分。最后简要介绍WebUITab与WebUIPage,并以Spark UI中的Environment页为例,分析了页面的展示流程。

— THE END —

原文发布于微信公众号 - 暴走大数据(zhouqiantanxi)

原文发表时间:2019-08-12

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券