知识中心

spark过节监控告警系统实现

2020-01-08 15:45:42 mimukeji

马上要过年了,大部分公司这个时候都不会再去谋求开新业务,而大数据工匠们,想要过好年,就要保证过年期间自己对自己的应用了如执掌。一般公司都会有轮值人员,至少要有春节应急预案,尤其是对于我们这些搞平台,或者线上应用的,应急预案更是必不可少。今天浪尖主要是分享一下关于在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.statusStore
statusStore.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.statusStore
statusStore.appSummary()statusStore.appSummary().numCompletedJobs
statusStore.appSummary().numCompletedStages

2.Job监控

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

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

image.png

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.statusStore
statusStore.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,分析源码得到的,框架细节只能如此获得。