Spark源码剖析

9. 启动测量系统MetricsSystem

MetricsSystem使用codahale提供的第三方测量仓库Metrics。MetricsSystem中有三个概念:

  • Instance:指定了谁在使用测量系统;
  • Source:指定了从哪里收集测量数据;
  • Sink:指定了从哪里输出测量数据;

Spark按照Instance的不同,区分为Master、Worker、Application、Driver和Executor。

Spark目前提供的Sink有ConsoleSink、CsvSink、JmxSink、MetricsServlet、GraphiteSink等。

Spark中使用MetricsServlet作为默认的Sink。

MetricsSystem在SparkEnv执行环境创建的过程中创建,代码如下:

MetricsSystem的启动代码如下:

MetricsSystem的启动过程包括以下步骤:

1) 注册Sources;

2) 注册Sinks;

3) 给Sinks增加Jetty的ServletContextHandler。

MetricsSystem启动完毕后,会遍历与Sinks有关的ServletContextHandler,并调用attachHandler将它们绑定到Spark UI上。代码如上图

9.1 注册Sources

registerSources方法用于注册Sources,告诉测量系统从哪里收集测量数据。注册Sources的过程分为以下步骤:

1) 从metricsConfig获取Driver的Properties,默认为创建MetricsSystem的过程中解析的{sink.servlet.class=org.apache.spark.metrics.sink.MetricsServlet,sink.servlet.path=/metrics/json}。

2) 用正则匹配Driver的Properties中以source.开头的属性。然后将属性中的Source发射得到的实例加入ArrayBuffer[Source]。

3) 将每个source的metricRegistry(也是MetricSet的子类型)注册到ConcurrentMap<String, Metric>metrics。

 

9.2 注册Sinks

registerSinks方法用于注册Sinks,即告诉测量系统MetricsSystem往哪里输出测量数据。注册Sinks的步骤如下:

1) 从Driver的Properties中用正则匹配以sink.开头的属性,如{sink.servlet.class=org.apache.spark.metrics.sink.MetricsServlet,sink.servlet.path=/metrics/json},将其转换为Map(servlet->{class=org.apache.spark.metrics.sink.MetricsServlet,path=/metrics/json})。

2) 将子属性class对应的类metricsServlet发射得到MetricsServlet实例。如果属性的key是serlvet,将其设置为metricsServlet;如果是Sink,则加入到ArrayBuffer[Sink]中。

 

9.3 给Sinks增加Jetty的ServletContextHandler

为了能够在SparkUI(网页)访问到测量数据,所以需要给Sinks增加Jetty的ServletContextHandler,这里主要用到MetricsSystem的getServletHandlers方法实现如下:

可以看到调用了metricsServlet的getHandlers,其实现如下:

最终生成处理/metrics/json请求的ServletContextHandler,而请求的真正处理由getMetricsSnapshot方法,利用fastjson解析。生成的ServletContextHandler通过SparkUI。最终我们可以使用以下这些地址来访问测量数据。

http://localhost:4040/metrics/applications/json
http://localhost:4040/metrics/json
http://localhost:4040/metrics/master/json

 

原文地址:https://www.cnblogs.com/swordfall/p/9317579.html