如何管理Spark Streaming消费Kafka的偏移量

Posted 2021-05-02 我是攻城师

前面的文章已经介绍了在spark streaming集成kafka时，如何处理其偏移量的问题，由于spark streaming自带的checkpoint弊端非常明显，所以一些对数据一致性要求比较高的项目里面，不建议采用其自带的checkpoint来做故障恢复。

在spark streaming1.3之后的版本支持direct kafka stream，这种策略更加完善，放弃了原来使用Kafka的高级API自动保存数据的偏移量，之后的版本采用Simple API也就是更加偏底层的api，我们既可以用checkpoint来容灾，也可以通过低级api来获取偏移量自己管理偏移量，这样以来无论是程序升级，还是故障重启，在框架端都可以做到Exact One准确一次的语义。

本篇文章，会再介绍下，如何手动管理kafka的offset，并给出具体的代码加以分析：

版本：

apache spark streaming2.1

apache kafka 0.9.0.0

手动管理offset的注意点：

（1）第一次项目启动的时候，因为zk里面没有偏移量，所以使用KafkaUtils直接创建InputStream，默认是从最新的偏移量开始消费，这一点可以控制。

（2）如果非第一次启动，zk里面已经存在偏移量，所以我们读取zk的偏移量，并把它传入到KafkaUtils中，从上次结束时的偏移量开始消费处理。

（3）在foreachRDD里面，对每一个批次的数据处理之后，再次更新存在zk里面的偏移量

注意上面的3个步骤，1和2只会加载一次，第3个步骤是每个批次里面都会执行一次。

下面看第一和第二个步骤的核心代码：

主要是针对第一次启动，和非首次启动做了不同的处理。

然后看下第三个步骤的代码：

主要是更新每个批次的偏移量到zk中。

例子已经上传到github中，有兴趣的同学可以参考这个链接：

https://github.com/qindongliang/streaming-offset-to-zk

后续文章会聊一下为了升级应用如何优雅的关闭的流程序，以及在kafka扩展分区时，上面的程序如何自动兼容。