SparkStreaming如何维护Kafka消息偏移量
SparkStreaming 维护 Kafka 消息偏移量据个人了解有两种方式
一、利用 SparkStreaming 自带的 Checkpoint 方法来维护
二、自己来编写维护 Kafka 消息偏移量的代码
首先说明下集群中的各组件版本
Hadoop-version： 2.7.1
Spark-version ： 1.6.2
zookeeper-version: 3.4.6
jdk-version： 1.8
maven-version： 3.3.3
kafka-version： 0.10.0
集群搭建方式 HDP(ambari)
首先说第一种 SparkStreaming 自带的 Checkpoint 方法，以下为代码示例
object Test {
def main(args: Array[String]): Unit = {
// TODO：创建检查点的位置可以设置为Hdfs如果程序重新启动spark程序会到此目录中检查并恢复
val checkpointDirectory ="";
// TODO：调用getOrCreate方法，这个方法入如果是第一次运行该作业
// TODO：没有checkpointDirectory该文件时
// TODO：将会重新创建一个StreamingContext，并从最新或是最老的偏移量处开始消费
val ssc: StreamingContext = StreamingContext.getOrCreate(checkpointDirectory, () => {
createStreaming(checkpointDirectory)
})
ssc.start()
ssc.awaitTermination()
}
// TODO：你的业务逻辑不应该写在main函数中，而是应该写在创建
// TODO：StreamingContext的方法中
def createStreaming(checkpointDirectory:String): StreamingContext = {
val brokerlist:String ="192.168.1.1:6667,192.168.1.1:6667"
val topic =Set("test")
val kafkaParams:Map[String,String] =Map[String,String]("metadata.broker.list" -> brokerlist)
val conf: SparkConf =new SparkConf()
.setMaster("local")
.setAppName(Test.getClass.getSimpleName)
val ssc: StreamingContext =new StreamingContext(conf,Seconds(3))
// TODO：在这里创建检查点，至于检查点在哪里创建，具体需要看个人的业务需求
ssc.checkpoint(checkpointDirectory)
val data: InputDStream[(String,String)] =
KafkaUtils.createDirectStream[String,String, StringDecoder, StringDecoder](ssc, kafkaParams, topic)
// TODO：以下代码就是你的业务逻辑了，这里只是循环遍历打印了Key，Value
data.foreachRDD(rdd => {
rdd.foreachPartition(partition => {
if (partition.isEmpty) {
for (tuple <- partition) {
println(tuple._1 + tuple._2)
}
}
})
})
ssc
}
}
说一下这个方法的缺点：
这个方法不适合总是需要迭代升级的应用，因为这个方法会在你建立检查点时将你的jar包信息以序列化的方式存在此目录中，
如果你的作业挂掉重新启动时，这时候是没有问题的，因为什么都没有改变。
但是在你的应用迭代升级时你的代码发生了变化，这是程序会发现其中的变化，你迭代升级后的版本将无法运行，就算是启动成功了，
运行的也还是迭代升级之前的代码。所还是以失败而告终！！
在Spark官方文档中给出了两个解决办法
第一个：老的作业不停机，新作业个老作业同时运行一段时间，这样是不安全的！！！
会导致数据重复消费，也有可能会发生数据丢失等问题
第二个：就是我要讲的自己维护消息偏移量
以下是自己维护消息偏移量代码示例
object Test{
def main(args: Array[String]): Unit = {
// TODO: Zookeeper 集群地址和端口
val zkHost:String ="192.168.1.1:2181,192.168.1.1:2181,192.168.1.1:2181"
// TODO：Kafka集群地址及端口
val brokerlist:String ="192.168.1.1:6667,192.168.1.1:6667,192.168.1.1:6667"
// TODO：指定消费 Kafka 主题
val topic:String ="test"
val kafkaParams:Map[String,String] =Map[String,String]("metadata.broker.list" -> brokerlist)
val conf: SparkConf =new SparkConf()
.setMaster("local[8]")
.setAppName(NGBoss_Dcc_Analysis.getClass.getSimpleName)
// TODO：获取一个StreamingContext对象
val ssc: StreamingContext =new StreamingContext(conf,Seconds(5))
// TODO：将获取的Zookeeper客户端
val zkClient =new ZkClient(zkHost)
var offsetRanges: Array[OffsetRange] =Array[OffsetRange]()
// TODO：设置在zookeeper中存储offset的路径
val topicDirs: ZKGroupTopicDirs =new ZKGroupTopicDirs("TEST_TOPIC_spark_streaming", topic)
val children = zkClient.countChildren(s"${topicDirs.consumerOffsetDir}")
var fromOffsets:Map[TopicAndPartition, Long] =Map()
var kafkaStream: InputDStream[(String, Array[Byte])] =null
if (children >0) {
//TODO:如果 zookeeper 中有保存 offset，我们会利用这个 offset 作为 kafkaStream 的起始位置
//TODO:如果保存过 offset，这里更好的做法，还应该和 kafka 上最小的 offset 做对比，不然会报 OutOfRange 的错误
for (i <-0 until children) {
val partitionOffset = zkClient.readData[String](s"${topicDirs.consumerOffsetDir}/${i}")
val tp =TopicAndPartition(topic, i)
//TODO:将不同 partition 对应的 offset 增加到 fromOffsets 中
fromOffsets += (tp -> partitionOffset.toLong)
}
//TODO:这个会将 kafka 的消息进行 transform，最终 kafak 的数据都会变成 (topic_name, message) 这样的 tuple
val messageHandler: (MessageAndMetadata[String, Array[Byte]]) => (String, Array[Byte]) = (mmd: MessageAndMetadata[String, Array[Byte]]) => (mmd.topic, mmd.message())
kafkaStream = KafkaUtils.createDirectStream[String, Array[Byte], StringDecoder, ByteDecoder, (String, Array[Byte])](ssc, kafkaParams, fromOffsets, messageHandler)
}else {
//TODO:如果未保存，根据 kafkaParam 的配置使用最新或者最旧的 offset
kafkaStream = KafkaUtils.createDirectStream[String, Array[Byte], StringDecoder, ByteDecoder] (ssc, kafkaParams,Set(topic))
}
kafkaStream.foreachRDD(rdd => {
if (!rdd.isEmpty()) {
rdd.foreachPartition(partition=> {
partition.foreach(tuple => {
println（tuple._1 + tuple._2）
})
})
offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
for (o <- offsetRanges) {
val zkPath:String =s"${topicDirs.consumerOffsetDir}/${o.partition}"
//TODO:将该 partition 的 offset 保存到 zookeeper
ZkUtils.updatePersistentPath(zkClient, zkPath, o.fromOffset.toString)
}
}
})
ssc.start()
ssc.awaitTermination()
}
}
第二种方法中offset可以存储在zookeeper中也可以存储在数据库中
自己维护offset是不会发生Checkpoint 中的问题，是我目前知道的最好的一个解决方案。
但是也有其中的缺点，如果从失败中恢复运行时不能获取到Key值，默认它的Key就是主题。
还有就是不能同时消费维护多个主题中的偏移量。
