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