作者：浪尖 原文連結
本文轉載自公衆号：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氛圍，歡迎釘釘掃碼加入！
對開源大資料和感興趣的同學可以加小編微信（下圖二維碼，備注“進群”）進入技術交流微信群。