上篇文章，讨论了在spark streaming中如何自己管理消费kafka的偏移量的方式，
这里在跟大家说一下这种方是的一些坑
由于想提高spark streaming程序的并行处理性能，于是需要增加kafka分区个数，，这里需要说下，在新版本spark streaming和kafka的集成中，按照官网的建议 spark streaming的executors的数量要和kafka的partition的个数保持相等，这样每一个executor处理一个kafka partition的数据，效率是最高的。如果executors的数量大于kafka的分区个数，其实多余的executors相当于是不会处理任何数据，这部分的进程其实是白白浪费性能。
如果executor的个数小于kafka partition的个数，那么其实有一些executors进程是需要处理多个partition分区的数据的，所以官网建议spark executors的进程数和kafka partition的个数要保持一致。
那么问题来了，如果想要提高spark streaming的并行处理性能，只能增加kafka的分区了，给kafka增加分区比较容易，直接执行一个命令即可，不过这里需要注意，kafka的分区只能增加不能减少，所以添加分区要考虑到底多少个才合适。
接下来我们便增加了kafka分区的数量，同时修改了spark streaming的executors的个数和kafka的分区个数一一对应，然后就启动了流程序，结果出现了比较诡异的问题，表现如下：
造几条测试数据打入kafka中，发现程序总是只能处理其中的一部分数据，而每次总有一些数据丢失。按理说代码没有任何改动，只是增加kafka的分区和spark streaming的executors的个数，应该不会出现问题才对，于是又重新测了原来的旧分区和程序，发现没有问题，经过对比发现问题只会出现在kafka新增分区后，然后出现这种丢数据的情况。然后和运维同学一起看了新增的kafka的分区的磁盘目录是否有数据落入，经查询发现新的分区确实已经有数据进入了，这就很奇怪了丢的数据到底是怎么丢的？
最后我又检查了我们自己保存的kafka的offset，发现里面的偏移量竟然没有新增kafka的分区的偏移量，至此，终于找到问题所在，也就是说，如果没有新增分区的偏移量，那么程序运行时是不会处理新增分区的数据，而我们新增的分区确确实实有数据落入了，这就是为啥前面说的诡异的丢失数据的原因，其实是因为新增kafka的分区的数据程序并没有处理过而这个原因正是我们的自己保存offset中没有记录新增分区的偏移量。
问题找到了，那么如何修复线上丢失的数据呢？
当时想了一个比较笨的方法，因为我们的kafka线上默认是保留7天的数据，旧分区的数据已经处理过，就是新增的分区数据没有处理，所以我们删除了已经处理过的旧的分区的数据，然后在业务流量底峰时期，重新启了流程序，让其从最早的数据开始消费处理，这样以来因为旧的分区被删除，只有新分区有数据，所以相当于是把丢失的那部分数据给修复了。修复完成后，又把程序停止，然后配置从最新的偏移量开始处理，这样偏移量里面就能识别到新增的分区，然后就继续正常处理即可。
注意这里面的删除kafka旧分区的数据，是一个比较危险的操作，它要求kafka的节点需要全部重启才能生效，所以除非特殊情况，不要使用这么危险的方式。
后来，仔细分析了我们使用的一个开源程序管理offset的源码，发现这个程序有一点bug，没有考虑到kafka新增分区的情况，也就是说如果你的kafka分区增加了，你的程序在重启后是识别不到新增的分区的，所以如果新增的分区还有数据进入，那么你的程序一定会丢数据，因为扩展kafka分区这个操作，并不常见，所以这个bug比较难易触发。
知道原因后，解决起来比较容易了，就是每次启动流程序前，对比一下当前我们自己保存的kafka的分区的个数和从zookeeper里面的存的topic的分区个数是否一致，如果不一致，就把新增的分区给添加到我们自己保存的信息中，并发偏移量初始化成0，这样以来在程序启动后，就会自动识别新增分区的数据。
所以，回过头来看上面的那个问题，最简单优雅的解决方法就是，直接手动修改我们自己的保存的kafka的分区偏移量信息，把新增的分区给加入进去，然后重启流程序即可。
个人理解：
场景：增加Kafka分区时的数据丢失
不过经过试验就算是自己维护偏移量也是会有相同的问题，因为你指定消费固定分区固定偏移量的数据，你没有指定消费新增分区里面的数据，所以也是同样会丢失新增分区内的数据的 。其实并不是开源框架的问题，更多的是问题考虑的不全面。
比如：如果你将offset存储在zookeeper中，当重启程序时，你从zookeeper中读取offset时你只读取了增加kafka分区之前的分区，也就是说增加之前有五个Kafka分区，增加分区后有七个Kafka分区,但是你的zookeeper中只保存了五个分区信息，所以在创建DStream的时候只消费zookeeper中保存的那五个分区中的数据，至于最后两个分区没有消费。所以会有数据丢失的现象。不过自己手动在zookeeper中添加后续添加的分区号，并将offset设置为0即可解决该问题
网友评论