ROI/Metrics 是什么职务？

ROI– refers to profitability

ROI/Metrics :投资回报率

这个职位要看一下是在什么性质的公司，

我觉得可能是那种投资公司或是顾问公司，应该是做投资回报率的分析统计之类的工作吧！

Flink Metrics指标采集方案

本文讨论的都是基于Flink On K8s场景下，该场景下存在几个特点，一是存在线上业务系统资源复用，二是调度节点存在"随机性"，对现有的Flink Metrics采集及使用姿势提出了新的要求：

Flink任务自动扩缩容，智能诊断场景依赖Metrics指标进行加工分析处理，现有Prometheus存储方案不再适合。

既有的指标采集需要先落本地，再由nodeexporter或lancer导出到目标存储，强依赖于Local环境，线上业务系统资源环境差异较大，扩容等维护成本较高，资源隔离性不够好。

期望在Flink On K8s场景下，Flink Metrics指标采集，能够不依赖于基础环境，对扩缩容友好，，支持指标采集及分析数据存储统一，降低指标维护使用成本，对Flink Metrics指标采集方案进行调研

2.1.1、 原理架构图如下

2.1.2、 配置方式

将flink-metrics-prometheus-1.14.3.jar 包放入到flink安装目录/lib下

修改flink-conf.yaml配置文件，设置属性

Example configuration:

metrics.reporter.promgateway.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter

metrics.reporter.promgateway.host: localhost

metrics.reporter.promgateway.port: 9091

metrics.reporter.promgateway.jobName: myJob

metrics.reporter.promgateway.randomJobNameSuffix: true

metrics.reporter.promgateway.deleteOnShutdown: false

metrics.reporter.promgateway.groupingKey: k1=v1;k2=v2

metrics.reporter.promgateway.interval: 60 SECONDS

2.2.1、原理架构图如下

2.2.2、配置方式

将flink-metrics-prometheus-1.14.3.jar 包放入到flink安装目录/lib下

修改flink-conf.yaml配置文件，设置属性

Example configuration:

metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter

metrics.reporter.prom.port: 9250-9260

2.3..1、原理架构图如下

2.3.2、配置方式

将flink-metrics-influxdb-1.14.3.jar 包放入到flink安装目录/lib下

修改flink-conf.yaml配置文件，设置属性

Example configuration:

metrics.reporter.influxdb.factory.class: org.apache.flink.metrics.influxdb.InfluxdbReporterFactory

metrics.reporter.influxdb.scheme: http

metrics.reporter.influxdb.host: localhost

metrics.reporter.influxdb.port: 8086

metrics.reporter.influxdb.db: flink

metrics.reporter.influxdb.username: flink-metrics

metrics.reporter.influxdb.password: qwerty

metrics.reporter.influxdb.retentionPolicy: one_hour

metrics.reporter.influxdb.consistency: ANY

metrics.reporter.influxdb.connectTimeout: 60000

metrics.reporter.influxdb.writeTimeout: 60000

metrics.reporter.influxdb.interval: 60 SECONDS

2.4.1、原理架构图如下

2.4.2、配置方式

将flink-metrics-jmx-1.14.3.jar 包放入到flink安装目录/lib下

修改flink-conf.yaml配置文件，设置属性

Example configuration:

metrics.reporter.jmx.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory

metrics.reporter.jmx.port: 9250-9260

2.5.1、配置方式

将flink-metrics-slf4j-1.14.3.jar 包放入到flink安装目录/lib下

修改flink-conf.yaml配置文件，设置属性

Example configuration:

metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory

metrics.reporter.slf4j.interval: 60 SECONDS

GraphiteReporter、StatsDReporter、DatadogHttpReporter

指标采集到Kafka后，将全量指标实时写入ClickHouse.

指标采集到Kafka后，将全量指标实时写入ClickHouse同时满足监控大盘需求及指标数据长期存储和二次加工分析，该方式优势指标数据源统一，任务大盘及告警

Spark Metrics

服务运行时将服务信息展示出来方便用户查看时服务易用性的重要组成部分。特别时对于分布式集群服务。

spark服务本身有提供获取应用信息对方法，方便用户查看应用信息。Spark服务提供对master，worker，driver，executor，Historyserver进程对运行展示。对于应用（driver／executor）进程，主要提供metric和restapi对访问方式以展示运行状态。

服务/进程通过Metric将自身运行信息展示出来。spark基于Coda Hale Metrics Library库展示。需要展示的信息通过配置source类，在运行时通过反射实例化并启动source进行收集。然后通过配置sink类，将信息sink到对应的平台。

以driver为例：driver进程启动metricSystem的流程：

SparkContext在初始化时调用 ： MetricsSystem.createMetricsSystem("driver", conf, securityManager)

然后等待ui启动后启动并绑定webui（executor则是初始化后直接启动）

metricsSystem.start()

metricsSystem.getServletHandlers.foreach(handler = ui.foreach(_.attachHandler(handler)))

创建MetricConfig， val metricsConfig = new MetricsConfig(conf)

初始化MetricConfig，首先设置默认的属性信息：

prop.setProperty("*.sink.servlet.class","org.apache.spark.metrics.sink.MetricsServlet")

prop.setProperty("*.sink.servlet.path","/metrics/json")

prop.setProperty("master.sink.servlet.path","/metrics/master/json")

prop.setProperty("applications.sink.servlet.path","/metrics/applications/json")

加载conf/metric.properties文件或者通过spark.metrics.conf制定的文件。读取相关配置，metricsConfig.initialize()

在启动metricSystem时，则会注册并启动source和sink

registerSources()

registerSinks()

sinks.foreach(_.start)

默认启动对source如下：

可配置的source如下：

配置方法：修改$SPARK_HOME/conf目录下的metrics.properties文件：

默认相关source已经统计在列。可添加source为jvmsource。添加之后则相关进程的jvm信息会被收集。配置方法

添加如下行：

driver.source.jvm.class=org.apache.spark.metrics.source.JvmSource

executor.source.jvm.class=org.apache.spark.metrics.source.JvmSource

或者*.source.jvm.class=org.apache.spark.metrics.source.JvmSource

source信息的获取比较简单，以DAGSchedulerSource的runningStages为例，直接计算dagscheduler的runningStages大小即可。override def getValue: Int = dagScheduler.runningStages.size

通过这些收集的信息可以看到，主要是方便查看运行状态，并非提供用来监控和管理应用

Metric信息展示方法：

收集的目的是方便展示，展示的方法是sink。

常用的sink如下：

a) metricserverlet

spark默认的sink为metricsserverlet，通过driver服务启动的webui绑定，然后展示出来。ip:4040/metrics/json(ip位driver节点的ip)展示：由于executor服务没有相关ui，无法展示metricsource的信息。 下图是配置过JVMsource后，通过driver节点的看到的metric信息。

b) CSV方式(将进程的source信息，写入到csv文件,各进程打印至进程节点的相关目录下，每分钟打印一次)：

*.sink.csv.class=org.apache.spark.metrics.sink.CsvSink

*.sink.csv.period=1

*.sink.csv.directory=/tmp/

c) console方式（将进程的source信息写入到console/stdout

,输出到进程的stdout）：

*.sink.console.class=org.apache.spark.metrics.sink.ConsoleSink

*.sink.console.period=20

*.sink.console.unit=seconds

d) slf4j方式（直接在运行日志中查看）：

*.sink.slf4j.class=org.apache.spark.metrics.sink.Slf4jSink

*.sink.slf4j.period=10

*.sink.slf4j.unit=seconds

e) JMX方式（此情况下，相关端口要经过规划，不同的pap使用不同的端口，对于一个app来说，只能在一个节点启动一个executor，否则会有端口冲突）：

executor.sink.jmx.class=org.apache.spark.metrics.sink.JmxSink

JMX方式在配置后，需要在driver/executor启动jmx服务。 可通过启动应用时添加如下操作实现--conf "spark.driver.extraJavaOptions=-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=8090 -Dcom.sun.management.jmxremote.rmi.port=8001 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false --conf "spark.executor.extraJavaOptions=-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=8002 -Dcom.sun.management.jmxremote.rmi.port=8003 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false"

可通过jconsole工具链接至对应driver进程所在ip和端口查看jmx信息。

除例metrics之外，用户还可以通过restApi接口查看应用运行信息。可以查询的信息如下（参见 ）：

运行中的应用：通过driver进程查看：

ip:port/api/v1/....

其中Ip为driver所在节点ip，端口为4040. 如果一个节点运行多个driver，端口会以此累加至4040，4041，4042 . 如：10.1.236.65:4041/api/v1/applications/application_1512542119073_0229/storage/rdd/23(on yarn 模式会自动跳转至如下页面)

对于运行完的应用，可通过jobhistory服务查看

此场景下，需要提交应用时打开eventlog记录功能

打开方法在应用的spark-defaults.conf中添加如下配置spark.eventLog.enabled为true，spark.eventLog.dir为hdfs:///spark-history 。

其中/spark-history可配置，需要和jobhistory进程的路径配置一致 ，该路径可通过historyserver页面查看。

ip:port/api/v1/....(其中Ip为spark服务的jobhistory进程所在节点ip，默认端口为18080). 可通过如下方式访问：

Spark作为计算引擎，对于大数据集群来说，作为客户端向Yarn提交应用来完成数据的分析。所使用的资源一般在yarn控制之下。其应用场景并非作为服务端为其他组件提供服务。其所提供的信息通常是针对app级别，如job，stage，task等信息。一般的信息监控需求均可通过其ui页面查看。对于一些应用的运行情况，可通过restapi获取和分析。

metrics-server采集数据失败问题排查

按照文档 《Kubernetes heapster监控插件安装》 和 《Kubernetes Metrics Server安装》 完成Heapster和Metrics Server之后，想尝试通过 kubectl top nodes 查看安装效果，结果抛出下面的错误：

我用下面一张图简单描述了一下APIService、Service、Pod、kubelet四者的关系，APIService负责对外提供服务，Pod访问个节点的kubelet抓取节点上各种指标数据，然后提供给APIService。

所以，我首先查看了一下APIService的状态，看APIService是否正常。通过下面的命令查看 kube-system/metrics-server APIService 的状态，可以看到有下面的报错信息：

从上面信息可以看到，APIService访问后端服务无响应，所以我再去查看Pod的日志：

从错误信息可以看到： metrics-server在调用各节点的kubelet获取对应节点上的指标信息时，指标数据没有有效的时间戳 。

根本原因是： 某些pod或node没有正常运行 ，如下图所示。只要保证pod或node运行正常后，这些错误信息就会消失。

然而，这个错误并不是根本原因，在纠结了两天之后，我突然想到，既然https请求无法访问，是不是和代理有关系。

我之前在 /etc/kubernetes/manifests/kube-apiserver.yaml 中加了一个https代理：

删掉这个代理问题就解决了，如果不能删除这个代理，也可以配置 no_proxy 环境变量，把10.108.78.78加上。

keras 自定义 metrics

在 keras 中操作的均为 Tensor 对象，因此，需要定义操作 Tensor 的函数来操作所有输出结果，定义好函数之后，直接将其放在 model.compile 函数 metrics 中即可生效：

使用方法如下：

custom metrics for binary classification in Keras

metrics.classification_report函数记录

机器学习/深度学习中，我们经常使用sklearn包中的metrics.classification_report来输出评价指标。本文主要是通过示例方式来记录该函数的常见输入与输出的含义。

示例1

其中，

accuracy 表示准确率，也即正确预测样本量与总样本量的比值，即9/13=0.69

macro avg 表示宏平均，表示所有类别对应指标的平均值，即

      precision = (1.0+0.67+0.5+0.67+1.0)/5=0.77

      recall = (0.67+0.67+0.67+1.0+0.5)/5=0.70

      f1-score = (0.8+0.67+0.57+0.8+0.67)/5=0.70

weighted avg 表示带权重平均，表示类别样本占总样本的比重与对应指标的乘积的累加和，即

      precision = 1.0*3/13 + 0.67*3/13 + 0.5*3/13 + 0.67*2/13 + 1.0*2/13=0.76

      recall = 0.67*3/13 + 0.67*3/13 + 0.67*3/13 + 1.0*2/13 + 0.5*2/13=0.69

      f1-score = 0.8*3/13 + 0.67*3/13 + 0.57*3/13 + 0.8*2/13 + 0.67*2/13=0.70

示例2

示例1跟示例2的区别是示例2加入了 target_names 参数，该参数的主要作用是将实际的类别与输出id对应起来。

示例3

对比示例2跟示例3，我们可以看到参数 target_names 中元素的顺序与输出id的大小顺序相同。

示例4

对于 classification_report ，我们通常会看到如下的输出

对比示例1跟示例4两个 classification_report 函数的输出，我们可以看到，示例1输出的是一个 accuracy ，外加两个平均值，而示例4输出的是四个平均值。

造成这种不同的原因是示例4是多标签分类，而示例1是一个单标签分类。

关于示例4输出的四个平均值，我们看看官方的解释

其中， micro avg 、 macro avg 和 weighted avg 针对的对象都是label，而 samples avg 针对的对象则是instance。

label表示示例4中值为1的三个样本中的元素，而instance表示实际的三个样本，具体不同可参考下面四个平均值的计算过程。

说明： classification_report 函数的输出结果标签中0、1、2、3、4表示的是每个样本的5列，每列代表一个标签。因此，对于标签0，表示样本中元素处于第一列的1；对于标签1，表示样本中元素处于第一列的1；以此类推。

micro avg 表示微平均，表示所有类别中预测正确量与总样本量的比值，即

      precision = (2+2+1+1) / 8 = 0.750

      recall = (2+2+1+1) / 9 = 0.667

      f1-score = 2*precision*recall/(precision+recall) = 0.706

macro avg 表示宏平均，表示所有类别对应指标的平均值，即

      precision = (1.0+1.0+0.5+1.0+0.0)/5 = 0.700

      recall = (1.0+1.0+0.5+1.0+0.0)/5 = 0.700

      f1-score = (1.0+1.0+0.5+1.0+0.0)/5 = 0.700

weighted avg 表示带权重平均，表示类别样本占总样本的比重与对应指标的乘积的累加和，即

      precision = 1.0*2/9 + 1.0*2/9 + 0.5*2/9 + 1.0*1/9 + 0.0*2/9 = 0.667

      recall = 1.0*2/9 + 1.0*2/9 + 0.5*2/9 + 1.0*1/9 + 0.0*2/9 = 0.667

      f1-score = 1.0*2/9 + 1.0*2/9 + 0.5*2/9 + 1.0*1/9 + 0.0*2/9 = 0.667

samples avg 表示带权重平均，表示类别样本占总样本的比重与对应指标的乘积的累加和，即

      precision = (1/2 + 2/3 + 3/3) / 3 = 0.722

      其中，1/2表示第1行中，标签值为1的预测准确率；2/3表示第二行中标签值为1的预测准确率，以此类推

      recall = (1/2 + 2/3 + 3/4) / 3 = 0.639

      f1-score = ((2*(1/2)*(1/2))/(1/2+1/2) + (2*(2/3)*(2/3))/(2/3+2/3) + (2*(3/3)*(3/4))/(3/3+3/4)) / 3 = 0.675

      其中，f1-score的计算过程中，分别为三个样本中标签值为1的f1值的平均值。

至此， classification_report 函数的常见输出结果介绍完成了，在此做个记录，方便自己与他人后续查阅。