Spark度量系统相关讲解
Spark的Metrics System的度量系统，有两个部分组成：source，sink，创建的时候需要制定instance。度量系统会周期的将source的指标数据被sink周期性的拉去，sink可以有很多。
Instance代表着使用度量系统的角色。在spark内部，目前master，worker，Executor，client driver，这些角色都会因为要去做监控而创建使用度量系统。目前，spark内部实现的instance有：master，worker，Executor，Driver，Applications。
Source指定定义了如何去收取度量指标。目前，已经存在以下两种source：
1.Spark内部的source，比如MasterSource，WorkerSource，ExecutorSource，
DAGSchedulerSource，BlockManagerSource，ApplicationSource。这些source会收集spark内部部件的状态。这些source都跟instance相关，在创建度量系统的时候会被加入。
2.公共的source，比如JVMSource，收集的是更加底层的状态，可以用配置文件配置并且是通过反射机制加载的。
Sink定义了度量指标数据输出的位置。同时可以共存很多sinks，指标数据会发给所有的sinks。
Source和sink的绑定
def start() {
require(!running, "Attempting to start a MetricsSystem that is already running")
running = true
registerSources()
registerSinks()
sinks.foreach(_.start)
}指标配置的格式如下：
[instance].[sink|source].[name].[options] = xxxx
[instance]可以是master，worker，executor,driver,applications.配置了就意味着只有指定的instance由此属性。可以粗犷的用*代替instance name，这就意味着所有的instance都将由此属性。
[sink|source].代表着该属性是source还是sink。只能是二选一。
[name]指定sink或者source的名字。
[options]指定sink或者source的属性
具体例子如下：
## Examples
# Enable JmxSink for all instances by class name
#*.sink.jmx.class=org.apache.spark.metrics.sink.JmxSink
# Enable ConsoleSink for all instances by class name
#*.sink.console.class=org.apache.spark.metrics.sink.ConsoleSink
# Polling period for ConsoleSink
#*.sink.console.period=10
#*.sink.console.unit=seconds
# Master instance overlap polling period
#master.sink.console.period=15
#master.sink.console.unit=seconds
# Enable CsvSink for all instances
#*.sink.csv.class=org.apache.spark.metrics.sink.CsvSink
# Polling period for CsvSink
#*.sink.csv.period=1
#*.sink.csv.unit=minutes
# Polling directory for CsvSink
#*.sink.csv.directory=/tmp/
# Worker instance overlap polling period
#worker.sink.csv.period=10
#worker.sink.csv.unit=minutes
# Enable Slf4jSink for all instances by class name
#*.sink.slf4j.class=org.apache.spark.metrics.sink.Slf4jSink
# Polling period for Slf4JSink
#*.sink.slf4j.period=1
#*.sink.slf4j.unit=minutes注意事项：
1,添加新的sink的时候，设置class option时需要是全名。
2,有些sink支持周期的拉去数据。最小拉去数据的周期是1秒钟。
3,有些特殊的属性支持通配符，例如：master.sink.console.period->*.sink.console.period
4,metrics.properties文件如果放在 ${SPARK_HOME}/conf目录下可以被自动加载
如果想自定义目录需要用-Dspark.metrics.conf=xxx，指定java属性配置的方式去指定。
5,MetricsServlet作为默认的sink，只支持，master，worker，client driver，可以通过发送http请求 /metrics/json,可以以json的格式获取所有已经注册的指标数据。
由于Spark生产中大部分运行于yarn上
Driver端的度量指标的请求方式
/proxy/application_1494227937369_0084/metrics/json
主要source源是：
StreamingSource，DAGSchedulerSource，BlockManagerSource，
ExecutorAllocationManagerSource
driver端的度量系统的初始化细节
在SparkContext里面
初始化度量系统
构建度量系统对象是在Sparkenv中做的
MetricsSystem.createMetricsSystem("driver", conf, securityManager)SparkContext只是引用了SparkEnv的对象
metricsSystem: MetricsSystem = if (_env != null) _env.metricsSystem else null启动度量系统并且绑定ServletHandler
// The metrics system for Driver need to be set spark.app.id to app ID.
// So it should start after we get app ID from the task scheduler and set spark.app.id.
metricsSystem.start()
// Attach the driver metrics servlet handler to the web ui after the metrics system is started.
metricsSystem.getServletHandlers.foreach(handler => ui.foreach(_.attachHandler(handler)))注册source
_env.metricsSystem.registerSource(_dagScheduler.metricsSource)
_env.metricsSystem.registerSource(new BlockManagerSource(_env.blockManager))
_executorAllocationManager.foreach { e =>
_env.metricsSystem.registerSource(e.executorAllocationManagerSource)
}Executor端的Source:
ExecutorSource
Executor端度量系统的初始化机启动
val metricsSystem = if (isDriver) {
// Don't start metrics system right now for Driver.
// We need to wait for the task scheduler to give us an app ID.
// Then we can start the metrics system.
MetricsSystem.createMetricsSystem("driver", conf, securityManager)
} else {
// We need to set the executor ID before the MetricsSystem is created because sources and
// sinks specified in the metrics configuration file will want to incorporate this executor's
// ID into the metrics they report.
conf.set("spark.executor.id", executorId)
val ms = MetricsSystem.createMetricsSystem("executor", conf, securityManager)
ms.start()
ms
}构建ExecutorSource并注册
private val executorSource = new ExecutorSource(threadPool, executorId)
if (!isLocal) {
env.metricsSystem.registerSource(executorSource)
env.blockManager.initialize(conf.getAppId)
}可以看到Executor端并没有绑定ServletHandler，故而无法通过http请求到度量指标。