记一次spark streaming+kafka 运行时间不稳定调优历程_streaming 任务如果长时间消费不到kafka 数据会挂么

问题现象

题主在去年8月份首次使用spark streaming进行流式计算的时候遇到的一个问题，即spark streaming读取kafka消息进行流式计算, 但是在数据量比较大的情况下总会出现一些batch的process time比较长，但是大多数batch能在较短的时间内完成，而且全部的batch运行时间呈两个极端分布，要么很长要么很短。

如上图，运行时间曲线出现多处尖峰，而我们期望的一般是连续平滑的曲线。

先说明题主这边的运行环境状况：集群搭建是基于HDP2.6.3版本，其中spark版本是2.11， kafka版本是0.10，其中spark、kafka、HDFS共享集群资源，业务诉求是每5分钟触发一次batch进行统计，需要在5分钟内计算完成，因为从运行图上看卡顿时间大约几十秒，并不会影响最终5分钟内运行完的要求，因此前期阶段只是断断续续在找原因并没有纠缠，但是近期接到项目要求统计1分钟粒度数据，那这个问题就极有可能影响到最后的运行是否按时完成，经过一段颇费周折的排查，终于找到原因，并将过程记录下来希望帮到遇到类似问题的同学。

排障调优历程

1. 单机差异排查
   通过streaming的web可看到运行的每个批的详细信息，我们注意到在运行时间上的批次里，基本都是少部分任务时间很长，大多数task还是很快的
   
   可以看到75%的task在1秒内执行完，但是Max最大的确实41秒，首先猜想的是机器性能差异，但是经过排查集群机器性能并无区别，且每台机器上执行的task比较均衡，基本上排除了单机性能差异问题，且在耗时较长的批次中，可以发现耗时长的task每次回出现在不同的机器上，因此也从侧面排除了机器性能差异问题。
2. 网络因素排查
   那么会不会是spark读取kafka网络问题呢
   
   从这张图可以看到，读取kafka全部具有数据本地性PROCESS_LOCAL，且集群内网万兆口互联，网络不应该成为瓶颈。
   且从这张图中可以排除数据量差异的原因以及GC耗时的原因。
3. 数据倾斜加盐排查
   既然数据量差异不是很大，接着对数据特征进行了筛查，通过最终打印多个批次的不同的key的数据量，我们发现并不存在时间长的批次里某些key的数据量突然增大的情况，且最后对数据量的较大的几个key加随机数分散处理，还是未能解决问题，也应证了不是数据倾斜的问题。
4. 单次消费数据量会话超时消费速率排查
   接着对kafka消费参数进行调节，主要有以下这么几个
   加大session.timeout.ms、减小max.poll.records，延长会话超时时间，减少每次从kafka拉取的数据量，参数经过多次调整，未解决问题。
5. checkpoint机制耗时排查
   也曾怀疑过是代码逻辑问题，但是整个代码逻辑很简单，就是读取数据进行写简单统计，复杂度不高，后面对代码的checkpoint进行了排查，因为网上有查到说checkpoint可能比较耗费性能，后面我们把checkpoint机制去除，发现checkpoint耗费的时间和process time不在一个数量级，也排除。
6. 执行器、驱动、堆内外内存配比排查
   为了验证不是内存不够的问题，我们把excutor memory、driver memory、spark.yarn.executor.memoryoverhead、spark.executor.memoryoverhead均增加了一倍，结果是内存足够，并不是内存的原因。
7. 集群重建问题复现排查
   后面题主怀疑过是集群的问题，在今年3月份有了一批压测机器，我们在全新的环境照着之前的模式再搭了一套HDP集群起来，但是发现还是有同样的问题，真是令人沮丧，一度换衣是spark的bug了。
8. kafka读取超时压缩排查
   后面再询问多为大数据圈内好友的经历后，建议修改了一个参数spark.streaming.kafka.consumer.poll.ms，将这个参数值改为了5000ms，神奇的现象出现了
   
   有两点，第一，耗时长的批次的process time不会“那么长”了，只比正常运行的batch多了3秒，第二，多了3秒的批次日志会报如下异常

19/06/28 20:00:05 WARN TaskSetManager: Lost task 0.0 in stage 510.0 (TID 54064, HDP-datanode02, executor 11): java.lang.AssertionError: assertion failed: Failed to get records for spark-executor-online-pingan-spark-testtt pingan 4 849148846 after polling for 5000
        at scala.Predef$.assert(Predef.scala:170)
        at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.get(CachedKafkaConsumer.scala:74)
        at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(KafkaRDD.scala:223)
        at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(KafkaRDD.scala:189)
        at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
        at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:462)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
        at org.apache.spark.sql.execution.columnar.InMemoryRelation$$anonfun$1$$anon$1.next(InMemoryRelation.scala:100)
        at org.apache.spark.sql.execution.columnar.InMemoryRelation$$anonfun$1$$anon$1.next(InMemoryRelation.scala:92)
        at org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:216)
        at org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1038)
        at org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1029)
        at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:969)
        at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1029)
        at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:760)
        at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:334)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:285)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
        at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
        at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
        at org.apache.spark.scheduler.Task.run(Task.scala:108)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:338)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

很明显，大致意思是5000ms内未从kafka中获取到任何数据信息，接下来会进行重试，原因找到了，之前未获取到消息会一直阻塞到大约延迟40秒拿到消息运行成功，现在是5秒内拿不到消息就重试，不长时间等待。

解决方案

我们可以先修改两个参数临时解决这个问题，第一，减小spark.streaming.kafka.consumer.poll.ms参数到3000ms以内，即3秒超时就重试，第二，将spark.task.maxFailures改为10，默认值是4，加大重试次数，修改完这两个参数后基本上解决了这个问题，多数批次在阻塞重连后都能很快读到消息并运行成功。但这只是临时解决方案，kafka集群不稳定是最根本的原因，最后我们还是建议将kafka集群和计算、存储集群分开部署，减少CPU、IO密集对消息队列带来的不稳定影响。