框架版本
spark2.1.0
kafka0.9.0.0
当使用sparkstreaming处理流式数据的时候,它的数据源搭档大部分都是Kafka,尤其是在互联网公司颇为常见。
当他们集成的时候我们需要重点考虑就是如果程序发生故障,或者升级重启,或者集群宕机,它究竟能否做到数据不丢不重呢?
也就是通常我们所说的高可靠和稳定性,通常框架里面都带有不同层次的消息保证机制,一般来说有三种就是:
at most once 最多一次
at least once 最少一次
exactly once 准确一次
在storm里面是通过ack和Trident,在sparkstreaming里面,如果是1.3版本之前是通过Receiver方式读取kafka数据,1.3之后通过Direct Approach方式直接读取kafka的数据,直接分配每个Batch及RDD最新的Topic partition offset,任务运行后使用kafka的Simple Consumer API去获取那一段的offset的数据,这样的好处是避免了原来Receiver接受数据宕机带来的数据可靠性风险,相当于原来的数据是在内存中而现在的数据是在kafka的磁盘中,通过偏移量可随时再次消费数据,从而实现了数据的Exactly Once处理,此外还有个不同之处在于1.3之后,使用的checkpoint保存当前消费的kafka的offset,而之前用zk保存的,这就是今天这篇文章重点吐槽的地方。
在sparkstreaming如何做到数据不丢失呢?
(1)使用checkpoint
(2)自己维护kafka偏移量
checkpoint配合kafka能够在特定环境下保证不丢不重,注意为什么要加上特定环境呢,这里有一些坑,checkpoint是对sparkstreaming运行过程中的元数据和
每次rdds的数据状态保存到一个持久化系统中,当然这里面也包含了offset,一般是HDFS,S3,如果程序挂了,或者集群挂了,下次启动仍然能够从checkpoint中恢复,从而做到生产环境的7*24高可用。
但是checkpoint的最大的弊端在于,一旦你的流式程序代码或配置改变了,或者更新迭代新功能了,这个时候,你先停旧的sparkstreaming程序,然后新的程序打包编译后执行运行,会发现两种情况:
(1)启动报错,反序列化异常
(2)启动正常,但是运行的代码仍然是上一次的程序的代码。
为什么会出现上面的两种情况,这是因为checkpoint第一次持久化的时候会把整个相关的jar给序列化成一个二进制文件,每次重启都会从里面恢复,但是当你新的
程序打包之后序列化加载的仍然是旧的序列化文件,这就会导致报错或者依旧执行旧代码。有的同学可能会说,既然如此,直接把上次的checkpoint删除了,不就能启动了吗? 确实是能启动,但是一旦你删除了旧的checkpoint,新启动的程序,只能从kafka的smallest或者largest的偏移量消费,默认是从最新的,如果是最新的,而不是上一次程序停止的那个偏移量
就会导致有数据丢失,如果是老的,那么就会导致数据重复。不管怎么样搞,都有问题。
https://spark.apache.org/docs/2.1.0/streaming-programming-guide.html#upgrading-application-code
针对这种问题,spark官网给出了2种解决办法:
(1)旧的不停机,新的程序继续启动,两个程序并存一段时间消费。 评价:仍然有丢重复消费的可能
(2)停机的时候,记录下最后一次的偏移量,然后新恢复的程序读取这个偏移量继续工作,从而达到不丢消息。 评价:官网没有给出具体怎么操作,只是给了个思路
第二种思路是正确的,但还需要自己维护一个offset状态,这样以来checkpoint这个功能只能在程序写好之后不允许再次变动,但可以重启的情况保证高可靠。
但实际情况是大多数公司的代码都会频繁迭代和升级,与checkpoint刚好相悖,这样以来checkpoint的作用便显的有点没用了,既然还是需要自己维护offset状态,
那么不用checkpoint也罢,完全自己维护offset状态到zk中即可。所以果断弃用checkpoint,采用自己维护offset。其原理如下:
首次启动,先从zk中找是否有上次存储的偏移量,如果没有就从最新的消费,然后保存偏移量至zk中
如果从zk中找到了偏移量,那么就从指定的偏移量处开始消费处理,每个批处理处理完毕后,都会更新新的offset到zk中,
这样以来无论是程序故障,还是宕机,再次启动后都会从上次的消费的偏移量处继续开始消费,而且程序的升级或功能改动新版本的发布都能正常运行
并做到了消息不丢。
需要注意的是,虽然上游能够做到准确一次的消费,但是下游的落地存储输出,比如写入hbase,redis,mysql,es等等如果失败了,整条消息依旧会失败,这个完全要靠自己的设计了,要么记录log,针对特定数据记录,如果失败定期 重新打入kafka走程序恢复或者手动恢复。
或者设计存储的时候,有复合主键,把偏移量提前,就算重复消费,但主键一样,最终只会有一条数据落地,这个要分场景和具体业务结合使用了。
回到主题,自己维护kafka的offset状态,如何做? github上已经有大神贡献了,我们只需要拿过来稍加改动即可,使用自己维护的offset之后,就没有必要再使用
checkpoint,github连接如下,有兴趣的朋友可以了解下:
https://github.com/cpbaranwal/Spark-Streaming-DirectKafka-Examples/blob/master/src/main/scala/CustomDirectKafkaExample.scala
使用zk维护offset也是比较不错的选择,如果将checkpoint存储在HDFS上,每隔几秒都会向HDFS上进行一次写入操作而且大部分都是小文件,且不说写入性能怎么样,就小文件过多,对整个Hadoop集群都不太友好。因为只记录偏移量信息,所以数据量非常小,zk作为一个分布式高可靠的的内存文件系统,非常适合这种场景。
所有参考链接:
http://aseigneurin.github.io/
http://aseigneurin.github.io/2016/05/07/spark-kafka-achieving-zero-data-loss.html
http://why-not-learn-something.blogspot.jp/2016/08/upgrading-running-spark-streaming.html
http://www.binwang.me/2015-11-03-the-proper-way-to-use-spark-checkpoint.html
https://github.com/cpbaranwal/Spark-Streaming-DirectKafka-Examples/blob/master/src/main/scala/CustomDirectKafkaExample.scala
https://github.com/ippontech/spark-kafka-source
有什么问题可以扫码关注微信公众号:我是攻城师(woshigcs),在后台留言咨询。
技术债不能欠,健康债更不能欠, 求道之路,与君同行。
分享到:
相关推荐
spark batch checkpoint jar
Spark的checkpoint源码讲解
import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.dstream.ReceiverInputDStream object UpadateDemo { def main(args: Array[String]): Unit = { val conf =...
下面是使用过程中记录的一些心得和博客,感兴趣的朋友可以了解下:项目简介该项目提供了一个在使用spark streaming2.3+kafka1.3的版本集成时,手动存储偏移量到zookeeper中,因为自带的checkpoint弊端太多,不利于...
Spark SQL的内置函数、开窗函数、UDF、UDAF,Spark Streaming的Kafka Direct API、updateStateByKey、transform、滑动窗口、foreachRDD性能优化、与Spark SQL整合使用、持久化、checkpoint、容错与事务。 7、多个从...
本文档详细的描述了spark的checkpoint的操作,帮你快速的了解
GitHub上某位大牛JerryLead对Spark的理解,大量图示,生动形象,总共7个pdf,看完对spark的原理,运行机制以及后续性能调优有很大的帮助,这是第六个pdf,讲述了cache、checkpoint的实现、使用等常见问题
该项目提供了一个在使用spark streaming2.1+kafka0.9.0.0的版本集成时,手动存储偏移量到zookeeper中,因为自带的checkpoint弊端太多,不利于 项目升级发布,并修复了一些遇到的bug,例子中的代码已
9.SparkStreaming Checkpoint 10.消费Kafka偏移量管理 第六章、StructuredStreaming模块 1.StructuredStreaming 概述(核心设计和编程模型) 2.入门案例:WordCount 3.输入源InputSources 4.Streaming Query 设置 5....
关于oracle checkpoint 的 经典解释。
Spark-submit关于参数及部署模式的部分解析 GroupByKey VS ReduceByKey OrderedRDDFunctions那些事 高效使用mappartitions standalone模式下executor调度策略 Spark Sql源码阅读 Spark Sql源码阅读 hive on ...
CheckPoint防火墙Nat配置讲解
Checkpoint防火墙命令行维护手册.pdf
CheckPoint\Checkpoint防火墙命令行维护手册
CheckPoint 4800 CPU 资源分配 CheckPoint 4800 CPU 资源分配 CheckPoint 4800 CPU 资源分配
ORACLE中的checkpoint
checkpoint UTM实施指南
zabbix template checkpoint snmp v2. checkpoint 模板 for zabbix
Checkpoint 简单配置手册.pdf
CheckPoint 防火墙配置手册V2.0