spark过节监控告警系统实现

2021-09-07 05:44:06

作者：浪尖
原文链接
本文转载自公众号：Spark学习技巧

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

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

阿里巴巴开源大数据技术团队成立Apache Spark中国技术社区，定期推送精彩案例，技术专家直播，问答区近万人Spark技术同学在线提问答疑，只为营造纯粹的Spark氛围，欢迎钉钉扫码加入！

码农公寓