六、如何管理Spark Streaming消费Kafka的偏移量（一）

柔情只为你懂 2024-04-19 16:09 63阅读 0赞

本篇我们先从理论的角度聊聊在Spark Streaming集成Kafka时的offset状态如何管理。

*  **spark streaming 版本 2.1**
 *  **kafka 版本0.9.0.0**

### **1、spark streaming CheckPoint 管理offset** ###

在这之前，先重述下spark streaming里面管理偏移量的策略，**默认的spark streaming它自带管理的offset的方式是通过checkpoint来记录每个批次的状态持久化到HDFS中**，如果机器发生故障，或者程序故障停止，下次启动时候，仍然可以从checkpoint的目录中读取故障时候rdd的状态，便能接着上次处理的数据继续处理。

**缺点**

但checkpoint方式***最大的弊端是如果代码升级***，***新版本的jar不能复用旧版本的序列化状态***，导致两个版本***不能平滑过渡***，结果就是***要么丢数据，要么数据重复***，所以官网搞的这个东西，几乎没有人敢在生产环境运行非常重要的流式项目。

**解决办法**

所以比较通用的解决办法就是***自己写代码管理spark streaming集成kafka时的offset，自己写代码管理offset，***其实就是把每批次offset存储到一个外部的存储系统里面包括（Hbase，HDFS,Zookeeper，Kafka，DB等等），不管用什么存储系统， 都需要***考虑到三种时刻的offset的状态***，否则offset的状态不完整，就可能导致一些bug出现。

### **2、自己管理offset需要考虑的三种offset状态** ###

**场景一：第一次启动**

当一个新的spark streaming+kafka的流式项目第一次启动的时候，这个时候发现外部的存储系统并没有记录任何有关这个topic所有分区的偏移量，所以就从 KafkaUtils.createDirectStream直接创建InputStream流，**默认是从最新的偏移量消费**，如果是第一次其实最新和最旧的偏移量时相等的都是0，然后在以后的每个批次中都会把最新的offset给存储到外部存储系统中，不断的做更新。

**场景二：项目停止后再次启动**

当流式项目停止后再次启动，会首先从外部存储系统读取是否记录的有偏移量，如果有的话，就读取这个偏移量，然后把偏移量集合传入到KafkaUtils.createDirectStream中进行构建InputSteam，这样的话就可以接着上次停止后的偏移量继续处理，然后每个批次中仍然的不断更新外部存储系统的偏移量，这样以来就能够无缝衔接了，无论是故障停止还是升级应用，都是透明的处理。

**场景三：streaming 运行期间增加了kafka的分区个数**

对正在运行的一个spark streaming+kafka的流式项目，我们在程序运行期间增加了kafka的分区个数，

**注意：**

*  这个时候新增的分区是***不能被正在运行的流式项目感应到的***，如果想要程序能够识别新增的分区，那么spark streaming应用程序**必须得重启**。
 *  同时如果你还使用的是自己写代码管理的offset就千万要注意，**对已经存储的分区偏移量，也要把新增的分区插入进去**，

否则你运行的程序仍然读取的是原来的分区偏移量，这样就会丢失一部分数据。

**总结：如果自己管理kafka的偏移量，一定要注意上面的三个场景，如果考虑不全，就有可能出现诡异的问题。**