spark过节监控告警系统实现

首先要祝大家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高手