Spark Streaming 中使用 zookeeper 保存 offset 并重用

在 Spark Streaming 中消费 Kafka 数据的时候，有两种方式分别是 1）基于 Receiver-based 的 createStream 方法和 2）Direct Approach (No Receivers) 方式的 createDirectStream 方法，详细的可以参考 Spark Streaming + Kafka Integration Guide，但是第二种使用方式中 kafka 的 offset 是保存在 checkpoint 中的，如果程序重启的话，会丢失一部分数据，可以参考 Spark Kafka - Achieving zero data-loss。

本文主要讲在使用第二种消费方式（Direct Approach）的情况下，如何将 kafka 中的 offset 保存到 zookeeper 中，以及如何从 zookeeper 中读取已存在的 offset。

大致思想就是，在初始化 kafka stream 的时候，查看 zookeeper 中是否保存有 offset，有就从该 offset 进行读取，没有就从最新/旧进行读取。在消费 kafka 数据的同时，将每个 partition 的 offset 保存到 zookeeper 中进行备份，具体实现参考下面代码

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
37
38
39
40
41
42
43
44
45
46
val topic : String = "topic_name"   //消费的 topic 名字
   val topics : Set[String] = Set(topic)                    //创建 stream 时使用的 topic 名字集合
   val topicDirs = new ZKGroupTopicDirs("test_spark_streaming_group", topic)  //创建一个 ZKGroupTopicDirs 对象，对保存
   val zkTopicPath = s"${topicDirs.consumerOffsetDir}"          获取 zookeeper 中的路径，这里会变成 /consumers/test_spark_streaming_group/offsets/topic_name
   val zkClient = new ZkClient("10.4.232.77:2181")          //zookeeper 的host 和 ip，创建一个 client
   val children = zkClient.countChildren(s"${topicDirs.consumerOffsetDir}")     //查询该路径下是否字节点（默认有字节点为我们自己保存不同 partition 时生成的）
   var kafkaStream : InputDStream[(String, String)] = null   
   var fromOffsets: Map[TopicAndPartition, Long] = Map()   //如果 zookeeper 中有保存 offset，我们会利用这个 offset 作为 kafkaStream 的起始位置
   if (children > 0) {   //如果保存过 offset，这里更好的做法，还应该和  kafka 上最小的 offset 做对比，不然会报 OutOfRange 的错误
       for (i <- 0 until children) {
         val partitionOffset = zkClient.readData[String](s"${topicDirs.consumerOffsetDir}/${i}")
         val tp = TopicAndPartition(topic, i)
         fromOffsets += (tp -> partitionOffset.toLong)  //将不同 partition 对应的 offset 增加到 fromOffsets 中
         logInfo("@@@@@@ topic[" + topic + "] partition[" + i + "] offset[" + partitionOffset + "] @@@@@@")
       }
       val messageHandler = (mmd : MessageAndMetadata[String, String]) => (mmd.topic, mmd.message())  //这个会将 kafka 的消息进行 transform，最终 kafak 的数据都会变成 (topic_name, message) 这样的 tuple
       kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc, kafkaParam, fromOffsets, messageHandler)
   }
   else {
       kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParam, topics) //如果未保存，根据 kafkaParam 的配置使用最新或者最旧的 offset
   }
   var offsetRanges = Array[OffsetRange]()
   kafkaStream.transform{ rdd =>
     offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges //得到该 rdd 对应 kafka 的消息的 offset
     rdd
   }.map(msg => msg._2).foreachRDD { rdd =&gt;     
     for (o <- offsetRanges) {
       val zkPath = s"${topicDirs.consumerOffsetDir}/${o.partition}"
       ZkUtils.updatePersistentPath(zkClient, zkPath, o.fromOffset.toString)  //将该 partition 的 offset 保存到 zookeeper
       logInfo(s"@@@@@@ topic  ${o.topic}  partition ${o.partition}  fromoffset ${o.fromOffset}  untiloffset ${o.untilOffset} #######")
     }
     rdd.foreachPartition(
       message => {
         while(message.hasNext) {
           logInfo(s"@^_^@   [" + message.next() + "] @^_^@")
         }
       }
     )
   }

使用上面的代码，我们可以做到 Spark Streaming 程序从 Kafka 中读取数据是不丢失

欢迎阅读第二篇文章，解决 offset out of range 的问题

Comments

▲