最近工作有点忙,所以更新文章频率低了点,在这里给大家说声抱歉,前面已经写过在spark streaming中管理offset,但当时只知道怎么用,并不是很了解为何要那样用,最近一段时间又抽空看了一个github开源程序自己管理offset的源码,基本已经理解透彻了,当然这里面还包含了由于理解不透彻导致升级失败的一个案例,这个在下篇文章会分享出来。本篇我们先从理论的角度聊聊在Spark Streaming集成Kafka时的offset状态如何管理。
spark streaming 版本 2.1
kafka 版本0.9.0.0
在这之前,先重述下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出现。
场景一:
当一个新的spark streaming+kafka的流式项目第一次启动的时候,这个时候发现外部的存储系统并没有记录任何有关这个topic所有分区的偏移量,所以就从 KafkaUtils.createDirectStream直接创建InputStream流,默认是从最新的偏移量消费,如果是第一次其实最新和最旧的偏移量时相等的都是0,然后在以后的每个批次中都会把最新的offset给存储到外部存储系统中,不断的做更新。
场景二:
当流式项目停止后再次启动,会首先从外部存储系统读取是否记录的有偏移量,如果有的话,就读取这个偏移量,然后把偏移量集合传入到KafkaUtils.createDirectStream中进行构建InputSteam,这样的话就可以接着上次停止后的偏移量继续处理,然后每个批次中仍然的不断更新外部存储系统的偏移量,这样以来就能够无缝衔接了,无论是故障停止还是升级应用,都是透明的处理。
场景三:
对正在运行的一个spark streaming+kafka的流式项目,我们在程序运行期间增加了kafka的分区个数,请注意:这个时候新增的分区是不能被正在运行的流式项目感应到的,如果想要程序能够识别新增的分区,那么spark streaming应用程序必须得重启,同时如果你还使用的是自己写代码管理的offset就千万要注意,对已经存储的分区偏移量,也要把新增的分区插入进去,否则你运行的程序仍然读取的是原来的分区偏移量,这样就会丢失一部分数据。
总结:
如果自己管理kafka的偏移量,一定要注意上面的三个场景,如果考虑不全,就有可能出现诡异的问题。
有什么问题可以扫码关注微信公众号:我是攻城师(woshigcs),在后台留言咨询。 技术债不能欠,健康债更不能欠, 求道之路,与君同行。
分享到:
相关推荐
一个手动管理spark streaming集成kafka时的偏移量到zookeeper中的小项目
spark-streaming-kafka-0-8_2.11-2.4.0.jar
【资源说明】 1、该资源包括项目的全部源码,下载可以直接使用! 2、本项目适合作为计算机、数学、电子信息等专业的课程设计、期末大作业和毕设...基于spark streaming+kafka+hbase的日志统计分析系统源码+项目说明.zip
spark streaming 链接kafka必用包,欢迎大家下载与使用
<scala.version>2.10.5 <spark.version>1.6.2</spark.version> <jackson.version>2.4.3 <hbase.version>1.2.0 的pom文件
spark streaming kafka
spakr streaming的kafka依赖
spark-streaming-kafka-0-8-assembly_2.11-2.4.0.jar
sparkStreaming消费数据不丢失,sparkStreaming消费数据不丢失
spark-streaming-kafka-0-10_2.11-2.4.0-cdh6.1.1.jar
spark streaming 链接kafka的必用包,国内下载很慢,特意分享出来,欢迎大家下载,速度很快哦
spark+kafka 项目jar包
KafkaUtils所依赖的jar包,导入文件中KafkaUtils报错,需要导入spark-streaming-kafka_2.10-1.6.0.jar包
Scala代码积累之spark streaming kafka 数据存入到hive源码实例,Scala代码积累之spark streaming kafka 数据存入到hive源码实例。
spark3.0.0版本对接kafka数据源需要的jar包,最新的版本导致maven的阿里云仓库不能直接下载下来,所以需要手动导入jar包进行操作,有需要的朋友可以免费下载
在WeTest舆情项目中,需要对每天千万级的游戏评论信息进行词频统计,在生产者一端,我们将数据按照每天的拉取时间存入了Kafka当中,而在消费者一端,我们利用了sparkstreaming从kafka中不断拉取数据进行词频统计。...
flume+Logstash+Kafka+Spark Streaming进行实时日志处理分析【大数据】
使用pyspark的stream操作kafka时,需要用到的jar包
介绍一个使用Spark Streaming和Kafka进行实时数据处理的示例。通过该示例,您将了解到如何使用Spark Streaming和Kafka处理实时数据流,以及如何将处理后的数据保存到MySQL数据库中。示例涵盖了从环境搭建到代码实现...
本使用kafka,spark,hbase开发日志分析系统