目录
前言
我们已经在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中的标签页，如下图中最上面的一栏所示，名称也是一一对应的。
接下来，调用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等等）的表格，干扰比较少。其页面本身如下图所示。
首先来看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页为例，分析了页面的展示流程。
