Spark心跳存活-延迟处理-异常日志告警

1.目录大纲

spark异常日志告警
spark存活心跳告警
spark批次处理延迟告警

2.任务背景

2.1问题背景

生成中针对spark steaming的程序进行告警的设置，生产上线需要监控spark streaming的异常日志的告警，及时发现生产线中的异常问题，方便快速定位问题；同时对spark 通过心跳机制对spark的异常下线问题告警，方便运维人员及时通知运维人员跟进，恢复程序运行。

2.2 技术路线

1.基于以往的告警研究，监控日志中的告警规则，监控异常日志中的error关键词告警；
2.基于spark自身的streamingListener机制，添加日志心跳异常检测告警；

2.3 告警邮箱

告警登录邮件的地址为：账号：xxxxxx_wuhan@163.com 密码：xxxxxx

3.安装部署过程

3.1spark业务日志异常告警

对应的告警规则如下，spark-ml-logwarn-rules.yaml的告警规则文件如下，

（该规则主要表示，在spark的业务日志中，在1min的时间窗口中，只要检测到ERROR关键词，就告警并发送邮件）

#spike：在相同query_key条件下，前后两个timeframe范围内数据量相差比例超过spike_height。其中可以通过spike_type设置具体涨跌方向是- up,down,both 。还可以通过threshold_ref设置要求上一个周期数据量的下限，threshold_cur设置要求当前周期数据量的下限>，如果数据量不到下限，也不触发；
#flatline：timeframe 范围内，数据量小于threshold 阈值；
#new_term：fields字段新出现之前terms_window_size(默认30天)范围内最多的terms_size (默认50)个结果以外的数据；
#cardinality：在相同 query_key条件下，timeframe范围内cardinality_field的值超过 max_cardinality 或者低于min_cardinality
## (Required)
## Type of alert.
## the frequency rule type alerts when num_events events occur with timeframe time
##我配置的是frequency，这个需要两个条件满足，在相同 query_key条件下，timeframe 范围内有num_events个被过滤出来的异常
type: frequency#这个index 是指再kibana 里边的index，支持正则匹配，支持多个index，同时如果嫌麻烦直接* 也可以。
## (Required)
## Index to search, wildcard supported
index: logwaring-ml-fct-*# 二选一，cur：当前窗口，ref：参照窗口 后边会着重讲述
#threshold_cur: 3
# #threshold_ref: 5num_events: 1timeframe:seconds: 60# 必选的，衡论是否进行报警的一个警戒线
#spike_height: 1# 必选的，spike_type有三种类型，up，down，both
#spike_type: "both"#这个还是非常关键的地方，就是你希望程序的message里边出现了什么样的关键字就报警，这个其实就是elasticsearch 的query语句，支持 AND&OR等。
#filter:
#- query:
#    query_string:
#      query: "param1:OBJECTBOUNDARY"
#filter:
- query_string:query: "level:*ERROR*"#在邮件正文会显示你定义的alert_text
alert_text: "你好，请回复邮件，方昌坦"# Setup report smtp config
smtp_host: "smtp.163.com"
smtp_port: 25
smtp_ssl: False#SMTP auth
from_addr: navinfo_wuhan@163.com
email_reply_to: navinfo_wuhan@163.com
smtp_auth_file: /opt/elastalert/config/smtp_auth.yaml# (Required)
# # The alert is use when a match is found
alert:
- "email"# (required, email specific)
# # a list of email addresses to send alerts to
email:
- "xxxx_wuhan@163.com"

其中的索引logwaring-ml-fct-*即为对应spark streaming（maplearning）的日志索引，基本最会查询最近1min的日志情况；

3.2 spark的心跳监控告警

3.2.1 集成告警代码到spark streaming代码中

集成告警代码到maplearning和broker公用的代码框架中，主要是在DataFromKafka 类中，添加spark侦听器端口，jsc.addStreamingListener(new SparkMonitoringListener("xxxx",xxxx));

该部分代码已经集成到maplearning系统中,添加单独的一个类如下，不停的向kafka中发送心跳信息；

package com.navinfo.csu.infrastructure.dataAccess.kafka;import com.navinfo.csu.infrastructure.configuration.GetEnvConf;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.spark.streaming.scheduler.*;import java.util.Properties;/** 通过sparkListener的机制，可以监控心跳、延迟批次过多告警；* @author fangchangtan* @date 2020-07-29 15:34*/
public class SparkMonitoringListener implements StreamingListener {private String appName ;private static String kafkaIp ;private KafkaProducer<String,String> producer = null;private String heatbeart_topic;private int  durationInterval = Integer.MAX_VALUE;public SparkMonitoringListener(String appName,String kafkaIp){this.kafkaIp = kafkaIp;this.appName = appName;this.durationInterval = Integer.valueOf(GetEnvConf.getValue("durationstime"));this.heatbeart_topic = GetEnvConf.getValue("spark-ml-heartbeat-topic");Properties props = new Properties();props.put("bootstrap.servers", kafkaIp);props.put("acks", "1");props.put("retries", "10");//自动处理的间隔时间1秒props.put("auto.commit.interval.ms", "1000");//设置key和value序列化方式props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("producer.type","sync");producer = new KafkaProducer<>(props);System.out.println("[SparkMonitoringListener] init ready to send heatbeat messge to heatbeart_topic: " +heatbeart_topic+" finished !!! ");};/*** 当该spark程序启动的时候，出发执行*/@Overridepublic void onStreamingStarted(StreamingListenerStreamingStarted streamingListenerStreamingStarted) {}/**** @param streamingListenerReceiverStarted*/@Overridepublic void onReceiverStarted(StreamingListenerReceiverStarted streamingListenerReceiverStarted) {}@Overridepublic void onReceiverError(StreamingListenerReceiverError streamingListenerReceiverError) {}@Overridepublic void onReceiverStopped(StreamingListenerReceiverStopped streamingListenerReceiverStopped) {}/*** 当每一个批次（目前ML是1min）提交的时候，触发方法执行。* 这个方式只要ml在运行，就会一直触发执行，相当于心跳的功能。** @param batchSubmitted*/@Overridepublic void onBatchSubmitted(StreamingListenerBatchSubmitted batchSubmitted) {final BatchInfo batchInfo = batchSubmitted.batchInfo();Object start = batchInfo.processingStartTime();Object end = batchInfo.processingEndTime();Object batchTime = batchInfo.batchTime();long LBatchTime = Long.valueOf(batchTime.toString().replaceAll(" ms",""));Object numRecords = batchInfo.numRecords();String heatbeatMsg = "onBatchSubmitted"+"@#"+start+"@#"+end+"@#"+LBatchTime+"@#"+numRecords;producer.send(new ProducerRecord<String, String>(heatbeart_topic, null, heatbeatMsg));}/*** 某一批次批次开始执行时候，触发该方法* @param batchStarted*/@Overridepublic void onBatchStarted(StreamingListenerBatchStarted batchStarted) {}/*** 某一批次数据结束执行时候，触发该方法* @param batchCompleted*/@Overridepublic void onBatchCompleted(StreamingListenerBatchCompleted batchCompleted) {final BatchInfo batchInfo = batchCompleted.batchInfo();Object SchedulingStartTime = batchInfo.processingStartTime().get();Object SchedulingEndTime = batchInfo.processingEndTime().get();String currentBatchTime = batchInfo.batchTime().toString().replace(" ms","");int currentNumRecords = Integer.valueOf(String.valueOf(batchInfo.numRecords()));final float totalDelay = Integer.valueOf(batchInfo.totalDelay().get().toString())/1000f;final float processingDelay = Integer.valueOf(batchInfo.processingDelay().get().toString()) /1000f;final int inComplitedBathNum = (int)(totalDelay / durationInterval);String listenerSparkMsg = "onBatchCompleted"+"@#"+ SchedulingStartTime +"@#"+SchedulingEndTime +"@#"+ currentBatchTime +"@#"+ currentNumRecords +"@#"+ processingDelay +"@#"+  totalDelay +"@#"+  inComplitedBathNum;producer.send(new ProducerRecord<String, String>(heatbeart_topic, null, listenerSparkMsg));}@Overridepublic void onOutputOperationStarted(StreamingListenerOutputOperationStarted streamingListenerOutputOperationStarted) {}@Overridepublic void onOutputOperationCompleted(StreamingListenerOutputOperationCompleted streamingListenerOutputOperationCompleted) {}
}

需要在原来的maplearning的部分添加的部分如下：

package com.navinfo.csu.infrastructure.dataAccess.kafka;import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;public class DataFromKafka {/*** @param jsc JavaSparkStreaming* @param ip kafka集群的ip地址* @param topic 接收kafka数据的topic,可以是一个topic，也可以是多个topic,多个topic应写为一个字符串，每个topic用逗号隔开* @return*/public static <T> JavaInputDStream<ConsumerRecord<String, T>> kafka(JavaStreamingContext jsc, String ip,String topic, String groupid, String valueType) {//###########################################################/**【心跳与批次监控模块】：* created by fang  2020/07/29* 1.技术路线：* StreamingListener接口提供类似spark ui的指标,提供心跳信号和批次指标到kafka，由logstash抓取后输出给elasticsearch，并通过elasticAlert监控es的index告警；* 2.功能描述：* ①可以将spark streaming按照提交的一个批次一次心跳（此处目前默认一般是1min，与spark自身的批次提交间隔同步），将心跳日志输出到elk告警。* 该心跳在spark的任务task阻塞的时候，仍然可以正常的输出心跳日志；因此，可以作为心跳的监控依据；* ②通过监听过多的批次延迟排队情况，或者对于单个批次执行时间过长超过阈值的情况，作出对应的告警；*/jsc.addStreamingListener(new SparkMonitoringListener("SparkStreaming",ip));//###########################################################if (groupid == null || groupid.equals("")) {groupid = "use_a_separate_group_id_for_each_stream";}Map<String, Object> kafkaParams = new HashMap<>();kafkaParams.put("bootstrap.servers", ip);kafkaParams.put("key.deserializer", StringDeserializer.class);if (valueType.equals(KafkaValueType.BYTEARRAY.toString())) {kafkaParams.put("value.deserializer", ByteArrayDeserializer.class);} else {kafkaParams.put("value.deserializer", StringDeserializer.class);}kafkaParams.put("group.id", groupid);kafkaParams.put("auto.offset.reset", "latest");kafkaParams.put("enable.auto.commit", true);kafkaParams.put("fetch.message.max.bytes", "300000000");kafkaParams.put("max.partition.fetch.bytes", "300000000");Collection<String> topics;if (topic.contains(",")) {topics = Arrays.asList(topic.split(","));} else {topics = Arrays.asList(topic);}// 从kafka订阅数据JavaInputDStream<ConsumerRecord<String, T>> stream = KafkaUtils.createDirectStream(jsc,LocationStrategies.PreferConsistent(), ConsumerStrategies.Subscribe(topics, kafkaParams));stream.count().print();return stream;}}

如题需要添加如下部分，其他部分保持不变；

       //###########################################################/**【心跳监控模块】：StreamingListener,提供心跳信号，输出心跳信息给elk监控告警；* created by fang  2020/07/29* 描述：该模块：可以将spark streaming按照提交的一个批次一次心跳（此处目前一般是1min），将心跳日志输出到elk告警。* 该心跳在spark的任务task阻塞的时候，让然可以正常的输出心跳日志；因此，可以作为心跳的监控依据；*/jsc.addStreamingListener(new SparkMonitorListener("MapLearning",ip));//###########################################################

3.2.2 同步修改启动的json配置文件

启动对应的maplearning程序之后，其心跳信息发送到kafka对应的topic(spark-ml-heatbeat-topic)中，然后通过logstash读取心跳信息到es中，然后通过elasticAlert的组件做告警处理；

启动的时候需要在maplearning程序的启动配置json文件中，添加"spark-ml-heartbeat-topic": "spark-ml-heartbeat",,这样程序可以传入心跳topic到程序中；

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-MmnqjmjA-1597127565632)(E:\Typora-fct\typora-workspace\typora-picture\image-20200730175950498.png)]

3.2.3 启动logstash接受心跳topic

使用logstash的配置如下：

#输入
input {kafka{bootstrap_servers => ["trial-kafkacluster-kafka.marathon.l4lb.thisdcos.directory:9292"]client_id => "client-spark-ml-heartbeat"group_id => "groupid-spark-ml-heartbeat"auto_offset_reset => "earliest"consumer_threads => 6decorate_events => truetopics => ["spark-ml-heartbeat"]type => "spark-listener"}
}#过滤
filter{if [type] == "spark-listener"{csv{columns=>["param0","param1","param2","param3","param4","param5","param6","param7"]separator=>"@#"}if [param0] == "onBatchCompleted"{mutate{convert => {"param6"=>"float"}convert => {"param7"=>"integer"}}}}}#输出
output {if [param0] == "onBatchSubmitted"{elasticsearch {hosts => ["cms-poctrial-productioncomponentselk-clusteres-nodex-xx.marathon.l4lb.thisdcos.directory:9202"]index => "spark-ml-heartbeat_%{+YYYY-MM-dd}"}#输出到控制台#stdout { codec => rubydebug }} else if [param0] == "onBatchCompleted" {elasticsearch {hosts => ["cms-poctrial-productioncomponentselk-clusteres-nodex-xx.marathon.l4lb.thisdcos.directory:9202"]index => "spark-ml-heartbeat_%{+YYYY-MM-dd}"}#输出到控制台#stdout { codec => rubydebug }}else {}
}

3.2.4 定义心跳告警规则

对应心跳的告警规则如下，其中的规则文件夹rules下的配置文件spark-ml-heartbeat-rules.yaml。

（该规则表示，连续相邻的两个1min大小的窗口中的心跳事件数量，变小超过一定值，开始发邮件报警！）

es_host: 172.19.32.xxxx
es_port: 9202#rule name 必须是独一的，不然会报错，这个定义完成之后，会成为报警邮件的标题
## (Required)
## Rule name, must be unique
#name: fct-spark-rule-name-Event-spike
name: maplearning-heatbeat-abnormal#############################################配置一种数据验证的方式，有 any，blacklist，whitelist，change，frequency，spike，flatline，new_term，cardinality
#any：只要有匹配就报警；
#blacklist：compare_key字段的内容匹配上 blacklist数组里任意内容；
#whitelist：compare_key字段的内容一个都没能匹配上whitelist数组里内容；
#change：在相同query_key条件下，compare_key字段的内容，在 timeframe范围内 发送变化；
#frequency：在相同 query_key条件下，timeframe 范围内有num_events个被过滤出 来的异常；
#spike：在相同query_key条件下，前后两个timeframe范围内数据量相差比例超过spike_height。其中可以通过spike_type设置具体涨跌方向是- up,down,both 。还可以通过threshold_ref设置要求上一个周期数据量的下限，threshold_cur设置要求当前周期数据量的下限>，如果数据量不到下限，也不触发；
#flatline：timeframe 范围内，数据量小于threshold 阈值；
#new_term：fields字段新出现之前terms_window_size(默认30天)范围内最多的terms_size (默认50)个结果以外的数据；
#cardinality：在相同 query_key条件下，timeframe范围内cardinality_field的值超过 max_cardinality 或者低于min_cardinality
## (Required)
## Type of alert.
## the frequency rule type alerts when num_events events occur with timeframe time
##我配置的是frequency，这个需要两个条件满足，在相同 query_key条件下，timeframe 范围内有num_events个被过滤出来的异常
type: spike#这个index 是指再kibana 里边的index，支持正则匹配，支持多个index，同时如果嫌麻烦直接* 也可以。
## (Required)
## Index to search, wildcard supported
index: spark-ml-heatbeat_*# 二选一，cur：当前窗口，ref：参照窗口 后边会着重讲述
#threshold_cur: 2
threshold_ref: 2#inum_events: 1
timeframe:minutes: 1# 必选的，衡论是否进行报警的一个警戒线
spike_height: 4# 必选的，spike_type有三种类型，up，down，both
spike_type: "down"#这个还是非常关键的地方，就是你希望程序的message里边出现了什么样的关键字就报警，这个其实就是elasticsearch 的query语句，支持 AND&OR等。
filter:
- query:query_string:query: "param0:onBatchSubmittedML"#########################################################
#邮件主题
alert_subject: "Alter {0} occurred at {1} {2}"
#主题中可以提供变量，变量值在此定义
alert_subject_args:
- _index
- "@timestamp"
- request
#在邮件正文会显示你定义的alert_text
alert_text: "管理员你好，你好，请回复邮件！ 检测到maplearning的心跳异常，请检查maplearning是否正常运行！"########################################################
# Setup report smtp config
smtp_host: smtp.163.com
smtp_port: 25
smtp_ssl: False#SMTP auth
from_addr: xxxxx_wuhan@163.com
email_reply_to: xxxxx_wuhan@163.com
smtp_auth_file: /opt/elastalert/config/smtp_auth.yaml# (Required)
# # The alert is use when a match is found
alert:
- "email"# (required, email specific)
# # a list of email addresses to send alerts to
email:
- "xxxxx_wuhan@163.com"

这样的规则可以保证，只有在maplearning程序在宕机的时候，esAlter可以及时发现心跳异常，作为告警邮件发送，其他时间不发送邮件通知；

3.3 spark批次延迟的告警

批次延迟的告警规则文件，vim spark-ml-delaywarn-rules.yaml，具体如下：

（该规则表示，在最近2分钟的时间窗口中，如果spark streaming未执行排队的批次数大于 1000批次时候，开始发邮件告警！）

es_host: 172.19.32.xxx
es_port: 9202#rule name 必须是独一的，不然会报错，这个定义完成之后，会成为报警邮件的标题
## (Required)
## Rule name, must be unique
name: spark-ml-delaywarn-frequency
#name: "警告：maplearning中监控到异常|错误！"#配置一种数据验证的方式，有 any，blacklist，whitelist，change，frequency，spike，flatline，new_term，cardinality
#any：只要有匹配就报警；
#blacklist：compare_key字段的内容匹配上 blacklist数组里任意内容；
#whitelist：compare_key字段的内容一个都没能匹配上whitelist数组里内容；
#change：在相同query_key条件下，compare_key字段的内容，在 timeframe范围内 发送变化；
#frequency：在相同 query_key条件下，timeframe 范围内有num_events个被过滤出 来的异常；
#spike：在相同query_key条件下，前后两个timeframe范围内数据量相差比例超过spike_height。其中可以通过spike_type设置具体涨跌方向是- up,down,both 。还可以通过threshold_ref设置要求上一个周期数据量的下限，threshold_cur设置要求当前周期数据量的下限>，如果数据量不到下限，也不触发；
#flatline：timeframe 范围内，数据量小于threshold 阈值；
#new_term：fields字段新出现之前terms_window_size(默认30天)范围内最多的terms_size (默认50)个结果以外的数据；
#cardinality：在相同 query_key条件下，timeframe范围内cardinality_field的值超过 max_cardinality 或者低于min_cardinality
## (Required)
## Type of alert.
## the frequency rule type alerts when num_events events occur with timeframe time
##我配置的是frequency，这个需要两个条件满足，在相同 query_key条件下，timeframe 范围内有num_events个被过滤出来的异常########################规则开始########################################
type: frequency#这个index 是指再kibana 里边的index，支持正则匹配，支持多个index，同时如果嫌麻烦直接* 也可以。
## (Required)
## Index to search, wildcard supported
index: spark-ml-heartbeat_*num_events: 1timeframe:seconds: 120filter:
- query_string:query: "param7: >1000"#########################规则结束##########################################
#在邮件正文会显示你定义的alert_text
alert_text: "你好，请回复邮件，方昌坦"# Setup report smtp config
smtp_host: "smtp.163.com"
smtp_port: 25
smtp_ssl: False#SMTP auth
from_addr: xxxxx_wuhan@163.com
email_reply_to: xxxxx_wuhan@163.com
smtp_auth_file: /opt/elastalert/config/smtp_auth.yaml# (Required)
# # The alert is use when a match is found
alert:
- "email"# (required, email specific)
# # a list of email addresses to send alerts to
email:
- "xxxxxx_wuhan@163.com"

注意：每次修改告警规则之后，都要重启elasticAlert的服务才能是规则生效！

**注释：**如图，心跳与延迟告警的logstash共用一个logstash服务。

4.总结

4.1 告警登录邮件的地址

账号：xxxxx_wuhan@163.com

密码：xxxxx

4.2 合理性

看看上面的过程是否合理；

(1)目前心跳不会因为task阻塞而影响spark心跳信息的发送，已经测试一直可以正常发送；

（2）spark的heartbeat信息不会与日志topic混在一起，单独一个topic，放置日志混合造成阻塞；

4.3改进意见

如果无问题，将按照上面的过程实施；