如何管理Spark Streaming消费Kafka的偏移量(一)

最近工作有点忙,所以更新文章频率低了点,在这里给大家说声抱歉,前面已经写过在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)

原文发表时间:2017-11-14

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏大数据技术学习

大数据-Hadoop小文件问题解决方案

HDFS中小文件是指文件size小于HDFS上block(dfs block size)大小的文件。大量的小文件会给Hadoop的扩展性和性能带来严重的影响。

2887
来自专栏Hadoop实操

如何使用StreamSets实时采集Kafka数据并写入Hive表

8522
来自专栏星汉技术

原 Spark的架构

3305
来自专栏牛肉圆粉不加葱

YARN 设计理念与基本架构

ResourceManager 是一个全局的资源管理器,负责整个集群的资源管理和分配。它主要由两个组件构成:调度器(Scheduler)和应用程序管理器(App...

991
来自专栏Spark学习技巧

Spark的调度系统

一,简介 Spark调度机制可以理解为两个层面的调度。首先,是Spark Application调度。也就是Spark应用程序在集群运行的调度,应用程序包括Dr...

2608
来自专栏LuckQI

Java大数据学习~Hadoop初识一了解其架构

913
来自专栏Hadoop实操

如何使用StreamSets从MySQL增量更新数据到Hive

温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看。 Fayson的github:https://github.com/fayson/cdhproje...

4.9K13
来自专栏个人分享

Hadoop的管理目录

1、NameNode的文件结构,NameNode会创建VERSION、edits、fsimage、fstime文件目录。其中dfs.name.dir属性是一个目...

3012
来自专栏简单聊聊Spark

Mapreduce实现原理总结

一.首先,我们从客户端,JobTracker,TaskTracker这个层面来分析整个mapreduce过程

2371
来自专栏Hadoop实操

CDH内存调拨过度警告分析

Hadoop应用程序或者Yarn的作业随机的出现OutOfMemory(OOM),在Cloudera Manager界面显示如下警告:

1.3K11

扫码关注云+社区

领取腾讯云代金券