spark streaming 应用程序 监控 邮件提醒

spark streaming应用程序，放到线上后，怎么监测spark streaming程序的阻塞状态，

虽然spark 提供了spark webUI去查看，但是作为开发人员总不能天天去看spark webUI页面吧，

去官网看，貌似可以通过请求spark 自带的jetty 服务器可以获取相关检测统计信息,

http://host:8088/proxy/application_1517299288666_7058/streaming/

返回的数据是要html页面，可以通过正则去解析需要的信息；但是这样子很不方便，能不能在spark streaming 发生阻塞的时候给我发送邮件提醒甚至是钉钉提醒呢？

这个方法比较笨，有没有更好的方法呢？跟踪spark StreamingContext源码：

spark streaming StreamingJobProgressListener 监控器

private val nextInputStreamId = new AtomicInteger(0)

private[streaming] var checkpointDir: String = {
if (isCheckpointPresent) {
sc.setCheckpointDir(cp_.checkpointDir)
cp_.checkpointDir
} else {
null
}
}

private[streaming] val checkpointDuration: Duration = {
if (isCheckpointPresent) cp_.checkpointDuration else graph.batchDuration
}

private[streaming] val scheduler = new JobScheduler(this)

private[streaming] val waiter = new ContextWaiter

private[streaming] val progressListener = new StreamingJobProgressListener(this)

private[streaming] val uiTab: Option[StreamingTab] =
if (conf.getBoolean("spark.ui.enabled", true)) {
Some(new StreamingTab(this))
} else {
None
}

StreamingJobProgressListener实现：

private[streaming] class StreamingJobProgressListener(ssc: StreamingContext)
extends StreamingListener with SparkListener {

private val waitingBatchUIData = new HashMap[Time, BatchUIData]
private val runningBatchUIData = new HashMap[Time, BatchUIData]
private val completedBatchUIData = new Queue[BatchUIData]
private val batchUIDataLimit = ssc.conf.getInt("spark.streaming.ui.retainedBatches", 1000)
private var totalCompletedBatches = 0L
private var totalReceivedRecords = 0L
private var totalProcessedRecords = 0L
private val receiverInfos = new HashMap[Int, ReceiverInfo]
......

然后我自己实现了一个Listener类，当发生阻塞的时候，可以发送邮件，以下实现比较简单

import org.apache.spark.streaming.scheduler._
import streaming.test.email.EmailSender
import org.slf4j._

class BJJListener(private val appName:String, private val duration: Int) extends StreamingListener{

private val logger = LoggerFactory.getLogger("BJJListener")

override def onReceiverStarted(receiverStarted: StreamingListenerReceiverStarted): Unit = {
super.onReceiverStarted(receiverStarted)
}

override def onReceiverError(receiverError: StreamingListenerReceiverError): Unit = super.onReceiverError(receiverError)

override def onReceiverStopped(receiverStopped: StreamingListenerReceiverStopped): Unit = super.onReceiverStopped(receiverStopped)

override def onBatchSubmitted(batchSubmitted: StreamingListenerBatchSubmitted): Unit = {
super.onBatchSubmitted(batchSubmitted)
val batchInfo = batchSubmitted.batchInfo
val batchTime = batchInfo.batchTime
logger.info("BJJListener  batchTime : ", batchTime)
}

override def onBatchStarted(batchStarted: StreamingListenerBatchStarted): Unit = {
val batchInfo = batchStarted.batchInfo
val processingStartTime = batchInfo.processingStartTime
logger.info("BJJListener  processingStartTime : ", processingStartTime)
}

override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = {
val batchInfo = batchCompleted.batchInfo
val processingStartTime = batchCompleted.batchInfo.processingStartTime

val processingEndTime = batchInfo.processingEndTime
val processingDelay = batchInfo.processingDelay
val totalDelay = batchInfo.totalDelay

if(totalDelay.get >= 6 * duration * 1000 && totalDelay.get >= 10 * duration * 1000){
val monitorTitle = s"spark streaming $appName 程序阻塞异常警告"
val monitorContent = s"BJJListener : processingStartTime -> ${processingStartTime.get}, processingEndTime -> ${processingEndTime.get} , " +
s"processingDelay -> ${processingDelay.get} , totalDelay -> ${totalDelay.get}, 请及时检查！"
EmailSender.sendMail(monitorTitle, monitorContent)
}
logger.info("BJJListener  processingEndTime : ", processingEndTime)
logger.info("BJJListener  processingDelay : ", processingDelay)
logger.info("BJJListener  totalDelay : ", totalDelay)
}

override def onOutputOperationStarted(outputOperationStarted: StreamingListenerOutputOperationStarted): Unit =
super.onOutputOperationStarted(outputOperationStarted)

override def onOutputOperationCompleted(outputOperationCompleted: StreamingListenerOutputOperationCompleted): Unit =
super.onOutputOperationCompleted(outputOperationCompleted)

}

通过可插拔的方式添加自己实现的listener

ssc.addStreamingListener(new BJJListener(appName, 10))

//spark streaming程序测试的例子：

import kafka.serializer.StringDecoder
import org.ansj.splitWord.analysis.NlpAnalysis
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
* Created by Amy on 2018/2/2.
*/
object test {
def main(args: Array[String]): Unit = {

System.setProperty("hadoop.home.dir", "D:\\mcyarn\\hadoop-common-2.2.0-bin-master")
Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
val appName = "spark Steaming test"
val conf = new SparkConf().setMaster("local[2]").setAppName("test")
val sc = new SparkContext(conf)
val ssc = new StreamingContext(sc, Seconds(10))

val brokerList = "localhost:9092"
val zookeeperConnect = "localhost:2181"
val groupId = "baasdf20180302"
val newsTopic = "test"

val kafkaParams = Map("metadata.broker.list" -> brokerList, "group.id" -> groupId,
"zookeeper.connect"->zookeeperConnect,
"auto.offset.reset" -> kafka.api.OffsetRequest.LargestTimeString)

val kafkaStream = KafkaUtils.createDirectStream[String,String,StringDecoder, StringDecoder](ssc, kafkaParams,
topics = Set(newsTopic)).map(_._1)

kafkaStream.foreachRDD(rdd=>{
if(!rdd.isEmpty()){

val rdds = rdd.union(rdd).union(rdd).union(rdd)
val transform = rdds.map(news=>{
if(news!=null){
val split = NlpAnalysis.parse(news).toStringWithOutNature(" ")
split
}else{
null
}
})

val wordCount = transform.map(word=>(word, 1)).reduceByKey(_+_)
wordCount.foreach(println)
println(rdd.count())
}
})

ssc.addStreamingListener(new BJJListener(appName, 10))

ssc.start()
ssc.awaitTermination()

}
}

阻塞到什么程序发送邮件，可以自己决定，可以发送邮件，也可以发送钉钉等，比较方便。

EmailSendInfo 类

import java.util.Properties;

public class EmailSendInfo {

// 发送邮件的服务器的IP和端口
private String mailServerHost;
private String mailServerPort = "25";
// 邮件发送者的地址
private String fromAddress;
// 邮件接收者的地址
private String toAddress;
// 登陆邮件发送服务器的用户名和密码
private String userName;
private String password;
// 是否需要身份验证
private boolean validate = false;
// 邮件主题
private String subject;
// 邮件的文本内容
private String content;
// 邮件附件的文件名
private String[] attachFileNames;

/**
* 获得邮件会话属性
*/
public Properties getProperties() {
Properties p = new Properties();
p.put("mail.smtp.host", this.mailServerHost);
p.put("mail.smtp.port", this.mailServerPort);
p.put("mail.smtp.auth", validate ? "true" : "false");
return p;
}

public String getMailServerHost() {
return mailServerHost;
}

/**
*
* @param mailServerHost
*/
public void setMailServerHost(String mailServerHost) {
this.mailServerHost = mailServerHost;
}

public String getMailServerPort() {
return mailServerPort;
}

/**
*
* @param mailServerPort
*/
public void setMailServerPort(String mailServerPort) {
this.mailServerPort = mailServerPort;
}

public boolean isValidate() {
return validate;
}

/**
*
* @param validate
*/
public void setValidate(boolean validate) {
this.validate = validate;
}

public String[] getAttachFileNames() {
return attachFileNames;
}

/**
*
* @param fileNames
*/
public void setAttachFileNames(String[] fileNames) {
this.attachFileNames = fileNames;
}

public String getFromAddress() {
return fromAddress;
}

/**
*
* @param fromAddress
*/
public void setFromAddress(String fromAddress) {
this.fromAddress = fromAddress;
}

public String getPassword() {
return password;
}

/**
*
* @param password
*/
public void setPassword(String password) {
this.password = password;
}

public String getToAddress() {
return toAddress;
}

/**
*
* @param toAddress
*/
public void setToAddress(String toAddress) {
this.toAddress = toAddress;
}

public String getUserName() {
return userName;
}

/**
*
* @param userName
*/
public void setUserName(String userName) {
this.userName = userName;
}

public String getSubject() {
return subject;
}

/**
*
* @param subject
*/
public void setSubject(String subject) {
this.subject = subject;
}

public String getContent() {
return content;
}

/**
*
* @param textContent
*/
public void setContent(String textContent) {
this.content = textContent;
}
}

EmailAuthenticator 类

import javax.mail.*;

public class EmailAuthenticator extends Authenticator{

private String userName;
private String password;

public EmailAuthenticator(){}

public EmailAuthenticator(String username, String password) {
this.userName = username;
this.password = password;
}

protected PasswordAuthentication getPasswordAuthentication(){
return new PasswordAuthentication(userName, password);
}

public String getUserName() {
return userName;
}

public void setUserName(String userName) {
this.userName = userName;
}

public String getPassword() {
return password;
}

public void setPassword(String password) {
this.password = password;
}

}

EmailSender

import javax.mail.*;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;
import java.util.Date;
import java.util.Properties;

public class EmailSender {

private static boolean sendTextMail(EmailSendInfo mailInfo) {
boolean sendStatus = false;//发送状态
// 判断是否需要身份认证
EmailAuthenticator authenticator = null;
Properties pro = mailInfo.getProperties();
if (mailInfo.isValidate()) {
// 如果需要身份认证，则创建一个密码验证器
authenticator = new EmailAuthenticator(mailInfo.getUserName(), mailInfo.getPassword());
}
// 根据邮件会话属性和密码验证器构造一个发送邮件的session
Session sendMailSession = Session.getInstance(pro, authenticator);
//【调试时使用】开启Session的debug模式
sendMailSession.setDebug(true);
try {
// 根据session创建一个邮件消息
MimeMessage mailMessage = new MimeMessage(sendMailSession);
// 创建邮件发送者地址
Address from = new InternetAddress(mailInfo.getFromAddress());
// 设置邮件消息的发送者
mailMessage.setFrom(from);
// 创建邮件的接收者地址，并设置到邮件消息中
Address to = new InternetAddress(mailInfo.getToAddress());
mailMessage.setRecipient(Message.RecipientType.TO,to);
// 设置邮件消息的主题
mailMessage.setSubject(mailInfo.getSubject(), "UTF-8");
// 设置邮件消息发送的时间
mailMessage.setSentDate(new Date());
// 设置邮件消息的主要内容
String mailContent = mailInfo.getContent();
mailMessage.setText(mailContent, "UTF-8");

// 发送邮件
Transport.send(mailMessage);
sendStatus = true;
} catch (MessagingException ex) {
ex.printStackTrace();
}
return sendStatus;
}

public static void sendMail(String monitorTitle, String monitorContent){
String fromaddr = "xxxx@yqzbw.com";
String toaddr = "xxxx@yqzbw.com";
String port = "25";
String host = "smtp.exmail.qq.com";
String userName = "xxxxxg@yqzbw.com";
String password = "12345678";

EmailSendInfo mailInfo = new EmailSendInfo();
mailInfo.setMailServerHost(host);
mailInfo.setValidate(true);
mailInfo.setUserName(userName);
mailInfo.setPassword(password);
mailInfo.setFromAddress(fromaddr);
mailInfo.setToAddress(toaddr);
mailInfo.setSubject(monitorTitle);
mailInfo.setContent(monitorContent);
//发送文体格式邮件
EmailSender.sendTextMail(mailInfo);
}

}

以上实现比较简单，有牛逼的大神，补充后，可以私信我