spark过节监控告警系统实现

news/2024/11/17 10:58:55/

首先要祝大家2020年快乐!

马上要过年了,大部分公司这个时候都不会再去谋求开新业务,而大数据工匠们,想要过好年,就要保证过年期间自己对自己的应用了如执掌。一般公司都会有轮值人员,至少要有春节应急预案,尤其是对于我们这些搞平台,或者线上应用的,应急预案更是必不可少。今天浪尖主要是分享一下关于在yarn上的spark 任务我们应该做哪些监控,如何监控。

 

Spark on yarn这种应用形态目前在企业中是最为常见的,对于这种spark的任务,浪尖觉得大家关心的指标大致有:app存活,spark streaming的job堆积情况,job运行状态及进度,stage运行进度,rdd缓存监控,内存监控等。

 

其实,春节最为重要的就是app存活了,春节期间各大应用应该都会有一部分数据增量,那么实际上就需要我们的程序能有一定的抗流量尖峰的能力,这个也很常见,因为正常的app都会有流量尖峰和低谷,你做一个实时应用程序,必须要去应对流量尖峰,也就是说你程序的处理能力正常要大于流量尖峰的,要是你的数据流量有历史信息,那么就简单了,只需要将spark streaming和flink的处理能力盖过流量最高值即可。当然,会有人说spark streaming 和flink不是有背压系统吗,短暂的流量尖峰可以抗住的呀,当然太短暂的几分钟的流量尖峰,而且你的任务对实时性要求不高,那是可以,否则不行。

1. App存活监控

企业中,很多时候spark的任务都是运行与yarn上的,这个时候可以通过yarn的客户端获取rm上运行 任务的状态。

Configuration conf = new YarnConfiguration();YarnClient yarnClient = YarnClient.createYarnClient();yarnClient.init(conf);yarnClient.start();try{   List<ApplicationReport> applications = yarnClient.getApplications(EnumSet.of(YarnApplicationState.RUNNING, YarnApplicationState.FINISHED));   System.out.println("ApplicationId ============> "+applications.get(0).getApplicationId());   System.out.println("name ============> "+applications.get(0).getName());   System.out.println("queue ============> "+applications.get(0).getQueue());   System.out.println("queue ============> "+applications.get(0).getUser());} catch(YarnException e) {   e.printStackTrace();} catch(IOException e) {   e.printStackTrace();}       yarnClient.stop();

这种api只适合,spark 和 MapReduce这两类应用,不适合flink。做过flink的应该都很容易理解吧,yarn上运行的flink任务显示,running,但是flink app内部的job却已经挂掉了,这种yarn的flink任务存活不适合,只能用RestClusterClient,具体浪尖在这里就不举例子了,本文主要是讲监控spark应用体系,后续会给出demo测试。

写个yarn的监控

对于这个APP的监控,还有更加细节的监控,比如executor数,内存,CPU等。获取指标的方法:

1.1 ApplicationInfo     

通过SparkContext对象的AppStatusStore对象获取ApplicationInfo

val statusStore = sparkContext.statusStorestatusStore.applicationinfo()

获取一个ApplicationInfo对象,然后主要包含以下schema

case class ApplicationInfo private[spark](    id: String,    name: String,    coresGranted: Option[Int],    maxCores: Option[Int],    coresPerExecutor: Option[Int],    memoryPerExecutorMB: Option[Int],    attempts: Seq[ApplicationAttemptInfo])

1.2 AppSummary

通过SparkContext对象的AppStatusStore对象 获取AppSummary

val statusStore = sparkContext.statusStorestatusStore.appSummary()
statusStore.appSummary().numCompletedJobsstatusStore.appSummary().numCompletedStages

2.Job监控

主要包括job的运行状态信息,spark streaming的job堆积情况。这个浪尖知识星球里面也分享过主要是自己实现一个StreamingListener,然后通过StreamingContext的实例对象注册到SparkListenerbus即可。

浪尖这里只会举一个就是spark streaming 数据量过大,导致batch不能及时处理而使得batch堆积,实际上就是active batch -1,针对这个给大家做个简单的案例,以供大家参考。

val waitingBatchUIData = new HashMap[Time, BatchUIData]ssc.addStreamingListener(new StreamingListener {  override def onStreamingStarted(streamingStarted: StreamingListenerStreamingStarted): Unit = println("started")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 = {    synchronized {      waitingBatchUIData(batchSubmitted.batchInfo.batchTime) =        BatchUIData(batchSubmitted.batchInfo)    }  }override def onBatchStarted(batchStarted: StreamingListenerBatchStarted): Unit =     waitingBatchUIData.remove(batchStarted.batchInfo.batchTime)    override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = super.onBatchCompleted(batchCompleted)override def onOutputOperationStarted(outputOperationStarted: StreamingListenerOutputOperationStarted): Unit = super.onOutputOperationStarted(outputOperationStarted)override def onOutputOperationCompleted(outputOperationCompleted: StreamingListenerOutputOperationCompleted): Unit = super.onOutputOperationCompleted(outputOperationCompleted)})

最终,我们使用waitingBatchUIData的大小,代表待处理的batch大小,比如待处理批次大于10,就告警,这个可以按照任务的重要程度和持续时间来设置一定的告警规则,避免误操作。

3. Stage监控

Stage的运行时间监控,这个重要度比较低。使用的主要API是statusStore.activeStages()得到的是一个Seq[v1.StageData] ,StageData可以包含的信息有:

class StageData private[spark](    val status: StageStatus,    val stageId: Int,    val attemptId: Int,    val numTasks: Int,    val numActiveTasks: Int,    val numCompleteTasks: Int,    val numFailedTasks: Int,    val numKilledTasks: Int,    val numCompletedIndices: Int,val executorRunTime: Long,    val executorCpuTime: Long,    val submissionTime: Option[Date],    val firstTaskLaunchedTime: Option[Date],    val completionTime: Option[Date],    val failureReason: Option[String],val inputBytes: Long,    val inputRecords: Long,    val outputBytes: Long,    val outputRecords: Long,    val shuffleReadBytes: Long,    val shuffleReadRecords: Long,    val shuffleWriteBytes: Long,    val shuffleWriteRecords: Long,    val memoryBytesSpilled: Long,    val diskBytesSpilled: Long,val name: String,    val description: Option[String],    val details: String,    val schedulingPool: String,val rddIds: Seq[Int],    val accumulatorUpdates: Seq[AccumulableInfo],    val tasks: Option[Map[Long, TaskData]],    val executorSummary: Option[Map[String, ExecutorStageSummary]],    val killedTasksSummary: Map[String, Int])

具体细节大家也可以详细测试哦。

 

4. RDD监控

这个其实大部分时间我们也是不关心的,主要是可以获取rdd相关的指标信息:

通过SparkContext对象的AppStatusStore

val statusStore = sparkContext.statusStorestatusStore.rddList()

可以获取一个Seq[v1.RDDStorageInfo]对象,可以获取的指标有:

class RDDStorageInfo private[spark](    val id: Int,    val name: String,    val numPartitions: Int,    val numCachedPartitions: Int,    val storageLevel: String,    val memoryUsed: Long,    val diskUsed: Long,    val dataDistribution: Option[Seq[RDDDataDistribution]],    val partitions: Option[Seq[RDDPartitionInfo]])
class RDDDataDistribution private[spark](    val address: String,    val memoryUsed: Long,    val memoryRemaining: Long,    val diskUsed: Long,    @JsonDeserialize(contentAs = classOf[JLong])    val onHeapMemoryUsed: Option[Long],    @JsonDeserialize(contentAs = classOf[JLong])    val offHeapMemoryUsed: Option[Long],    @JsonDeserialize(contentAs = classOf[JLong])    val onHeapMemoryRemaining: Option[Long],    @JsonDeserialize(contentAs = classOf[JLong])    val offHeapMemoryRemaining: Option[Long])
class RDDPartitionInfo private[spark](    val blockName: String,    val storageLevel: String,    val memoryUsed: Long,    val diskUsed: Long,    val executors: Seq[String])

其中,还有一些api大家自己也可以看看。

5. Rdd内存及缓存监控

主要是监控executor的内存使用情况,然后对一些高内存的任务能及时发现,然后积极排查问题。这个问题监控也比较奇葩,主要是监控RDD的内存和磁盘占用即可。对于缓存的rdd获取,只需要statusStore.rddList()获取的时候给定boolean参数true即可。获取之后依然是一个RDD列表,可以参考4,去进行一些计算展示。

 

6.Executor监控

关于内存的监控,除了存活监控之外,还有单个executor内存细节。Executor的注册,启动,挂掉都可以通过SparkListener来获取到,而单个executor内部的细节获取也还是通过SparkContext的一个内部变量,叫做SparkStatusTracker。

sc.statusTracker.getExecutorInfos

得到的是一个Array[SparkExecutorInfo],然后通过SparkExecutorInfo就可以获取细节信息:

private class SparkExecutorInfoImpl(    val host: String,    val port: Int,    val cacheSize: Long,    val numRunningTasks: Int,    val usedOnHeapStorageMemory: Long,    val usedOffHeapStorageMemory: Long,    val totalOnHeapStorageMemory: Long,    val totalOffHeapStorageMemory: Long)  extends SparkExecutorInfo

7.总结

浪尖常用的监控就是一app存活监控,二就是定义sparklistener,实现检测sparkstreaming 队列积压了。

总有粉丝文浪尖,如何发现这些细节的,当然是看源码的api,分析源码得到的,框架细节只能如此获得。

有问题可以到菜单栏里加浪尖微信哦。

推荐阅读:

Flink 在 字节跳动

flink sql使用中的一个问题

干货 | 如何成为大数据Spark高手


http://www.ppmy.cn/news/774622.html

相关文章

JVM下篇:性能监控与调优篇_05_分析GC日志_尚硅谷

文章目录 01-GC日志参数02-GC日志格式复习&#xff1a;GC分类哪些情况会触发Full GC&#xff1f; 不同GC类的GC细节GC日志分类Minor GCFull GC 日志结构剖析垃圾收集器GC前后情况GC时间 Minor GC 日志解析Full GC日志解析 03-GC日志分析工具GCeasy&#xff08;比较实用&#xf…

Window,Linux及应用程序监控

Windows CPU&#xff0c;内存&#xff0c;磁盘&#xff0c;网络&#xff0c;GPU等可以在任务管理器中的性能选项查看。 打开左下角的资源监视器可以具体到每个进程。 Perfmon工具 图像根据《大话Java性能调优》而截的图 然后点击性能监视器&#xff0c;然后点击添加按钮&…

zabbix如何选择适合的监控类型

zabbix如何选择适合的监控类型 zabbix提供十几种监控类型&#xff0c;包括&#xff1a;Zabbix agent, Simple checks, SNMP, Zabbix internal, IPMI, JMX monitoring等等&#xff0c;那我们应该如何选择呢&#xff1f;凉白开在此给大家一一作一个说明 zabbix监控类型 zabbix …

移动开发 | 移动端 APM 性能监控

移动互联网时代&#xff0c;各大互联网公司都已将自己的产品和服务全面移动化&#xff0c;各类新产品功能都会优先在移动APP上尝试。 应用性能作为影响用户体验最重要的因素&#xff0c;在开发过程中显得尤为重要。 1 用户网络环境不稳定&#xff0c;兼容性差&#xff1f; 2 终…

用于CPU性能SQL Server监视工具

CPU压力降低了服务器速度 (CPU pressure slowing down the server) This article is the sequel in a series about SQL Server monitoring tools and common performance issues. Before reading this piece, it advisable to read the previous two articles about monitorin…

【监控仪表系统】Grafana 中文入门教程 | 构建你的第一个仪表盘

Grafana 读音&#xff1a;/grəˈfɑːnˌɑː/ Grafana 中文入门教程 1. Grafana 是什么Grafana 支持的数据源 2. 什么情况下会用到 Grafana 或者监控仪表盘3. 安装和配置 Grafana4. Grfana 工作原理5. 搭建你的第一个仪表盘第 1 步 - 设置数据源第 2 步 - 导入 Dashboard第 3…

JVM性能监控及调优篇

一&#xff0c;概述 1&#xff0c;背景说明 1&#xff09;生产环境中的问题 生产环境发生了内存溢出该如何处理 生产环境应该给服务器分配多少内存合适&#xff1f; 如何对垃圾回收器的性能进行调优&#xff1f; 生产环境CPU负载飙高该如何处理&#xff1f; 生产环境应该…

可观测性就是对“监控”的包装?

持续坚持原创输出&#xff0c;点击蓝字关注我吧 作者&#xff1a;软件质量保障 知乎&#xff1a;https://www.zhihu.com/people/iloverain1024 微服务和分布式架构已成为构建服务应用程序的新模式。随着对软件可扩展性要求的提高&#xff0c;系统的复杂性和动态性也在不断提高。…