通过Spark listener实现Direct模式读取Kafaka数据

参考文章：

http://coolplayer.net/2016/11/30/spark-streaming-从kafka-拉数据如何保证数据不丢失/

https://github.com/jacksu/utils4s/blob/master/spark-knowledge/md/spark_streaming使用kafka保证数据零丢失.md?hmsr=toutiao.io&utm_medium=toutiao.io&utm_source=toutiao.io

为什么使用 direct 方式

Direct Approach VS Receiver-based Approach

因为按需拉数据，所以不存在缓冲区，就不用担心缓冲区把内存撑爆了。这个在Receiver-based Approach 就比较麻烦，你需要通过spark.streaming.blockInterval等参数来调整。
数据默认就被分布到了多个Executor上。Receiver-based Approach 你需要做特定的处理，才能让 Receiver分不到多个Executor上。
Receiver-based Approach 的方式，一旦你的Batch Processing 被delay了，或者被delay了很多个batch,那估计你的Spark Streaming程序离奔溃也就不远了。 Direct Approach (No Receivers) 则完全不会存在类似问题。就算你delay了很多个batch time,你内存中的数据只有这次处理的。
能保证exact once 语意, 图解参考

spark streaming 里面应该怎么做

spark 可以使用 Direct 和 Receiver-based 两种方式从kafka中取数据，显然我们应该选用Direct 方式
在处理过程中应该考虑如何 recovery, 这样我们需要把每个batch中的分区消费位置持久化存储在hdfs上，为了重启的时候可以从上次断掉的位置继续消费

实现思路如下

我们可以 override StreamingListener 的onBatchCompleted函数, 在每个 batch 处理完的时候保存一下当前处理的kafka 的 offset, offset 数组信息可以从InputInfoTracker中获取，

重启的时候可以从持久化目录里面获取最后消费的分区消费位置数组，然后设置一下 DirectKafkaInputDStream 的 currentOffsets 字段，就可以做到从上次断掉的位置继续消费

关键代码如下

streamingContext.addStreamingListener(new BatchStreamingListener(this))

class BatchStreamingListener(runtime: SparkStreamingRuntime) extends StreamingListener {override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = {operator.directKafkaRecoverSource.saveJobSate(time)}
}

val info =  ssc.scheduler.inputInfoTracker.getInfo(time)
info(jobName).metadata("offsets")

这里确保每一批数据消费完了之后，就持久化kafka中的分区offset 数组, 可以从 inputInfoTracker 获取每个batch 中处理的offset数组，写成一个文件，下面就是根据参数中设置的hdfs持久化路径上传到hdfs

val field = classOf[DirectKafkaInputDStream[_, _, _, _, _]].getDeclaredField("currentOffsets")
field.setAccessible(true)
field.set(dkid, state)

测试下面就简单了，在下次重启spark streaming 的时候遍历hdfs持久化路径中最末尾的文件，找到其中保存的kafka offset 数组，然后设置 DirectKafkaInputDStream 中的currentOffsets 字段，然后就可以做到从上次断掉的位置继续消费

下面模拟一个spark streaming 从kafka中消费数据的场景，由于spark streming 中是direct 的方式， offset的数组没有打入 zookeeper ，所以kafka 中自带的监控工具是失效的，这里有一个比较 trick的方法，你可以使用 bin/kafka-console-consumer.sh 工具消费同一个topic, 这样的话你就可以从kafka 的监控工具中看到消息写到哪个位置了，然后再观察你spark 里面持久化的位置数组，然后就可以确认是从上次断掉的位置继续消费还是从最新的位置消费

转载于:https://www.cnblogs.com/liugh/articles/6991993.html