spark-streaming与kafka的整合
Posted amddbb
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了spark-streaming与kafka的整合相关的知识,希望对你有一定的参考价值。
1. 概述
在2.x中,spark有两个用来与kafka整合的代码,版本代号为0.8和0.10,由于在0.8,kafka有两套消费者api,根据高级api得到了Receiver-based Approach,根据低级api得到了Direct Approach,而在0.10由于kafka只有一套消费者api了,所以也只有Direct Approach
2. Direct Approach
由于0.8与0.10的Direct Approach的实现差不多,这里使用0.10来进行描述。该途径由DirectKafkaInputDStream
实现
private[spark] class DirectKafkaInputDStream[K, V](
_ssc: StreamingContext,
locationStrategy: LocationStrategy,
consumerStrategy: ConsumerStrategy[K, V],
ppc: PerPartitionConfig
)
其中,LocationStrategy
表示的是分配topic partition的consumer到executor的策略,有3个取值
PreferConsistent
: 均匀的分布在executor中。原理是使用TopicPartition
的hash值来决定到哪个可用的executor上执行PreferBrokers
:偏向于将consumer分配到指定的topic partition所在的leader上;如果leader集中,会造成负载不均。如果可用的executor中没有leader所在的主机,使用PreferConsistent
PreferFixed
: 指定某些topic partition的consumer分配到指定的机器上,没有指定的topic partition使用PreferConsistent
ConsumerStrategy
用来指定消费的topic partition,也有三个子类来表示不同的策略:
Subscribe
:精确指定用来消费的topicSubscribePattern
:使用正则表达式来指定消费的topicAssign
: 精确指定用来消费的topic partition。上述2个都是消费topic的所有分区,这个可以选择性的消费部分分区
PerPartitionConfig
只有一个实现类DefaultPerPartitionConfig
,用来执行反压
,控制每个分区的消费速率
2.1 执行过程
分为2步,一是DirectKafkaInputDStream
计算每个分区及其消费开始的offset与结束的offset(不包含),二是将上述信息封装到KafkaRDD
中消费每个topic partition,一个topic partition对应一个task
2.2 常用的配置
配置分为两种,一种是传给ConsumerStrategy
,为Kafka Consumer相关的配置,还有一种是传给SparkConf
的。
传给ConsumerStrategy
的kafkaParams
变量名 | 描述 | 其它 |
---|---|---|
heartbeat.interval.ms | 客户端与broker的心跳间隔 | 不小于batchDuration ,在有window操作时,要不小于slideDuration
|
session.timeout.ms | 在该时间内,如果收不到客户端心跳,会移除该客户端 | 推荐3 * heartbeat.interval.ms,但要在范围(group.min.session.timeout.ms, group.max.session.timeout.ms]
|
metadata.broker.list | kafka集群的broker列表,用来获取整个集群拓扑,不用完整 | 等价于bootstrap.servers |
enable.auto.commit | 是否自动提交消费的offset | 在executor端,为false,通常也要设置为false |
auto.commit.interval.ms | 自动提交消费的offset的周期 | 默认为5s |
group.id | 消费组的id | 尽量设置 |
auto.offset.reset | 当指定消费的offset不存在时,采取的行为 | 在executor端为none |
receive.buffer.bytes | The size of the TCP receive buffer (SO_RCVBUF) to use when reading data | 默认65536,设置的时候要大于该值 |
注意上述的默认值要根据使用的版本来确定。
传给SparkConf
的
变量名 | 描述 | 其它 |
---|---|---|
spark.streaming.kafka.consumer.poll.ms | 消费者获取消息等待的最大时间 | 默认值spark.network.timeout |
spark.streaming.kafka.allowNonConsecutiveOffsets | 是否允许消费的offset不连续,通常都是连续的,除非中间的message被删掉了 | 默认值false |
spark.streaming.kafka.consumer.cache.enabled | 在executor端将消费指定topic partition的consumer缓存起来,这个是executor级别的缓存,如果没有使用动态资源分配,可以用来重用consumer | 默认值true |
spark.streaming.kafka.consumer.cache.maxCapacity | 每个executor上能缓存的consumer的最大值 | 默认值64 |
spark.streaming.kafka.consumer.cache.minCapacity | 每个executor上能缓存的consumer的最大值 | 默认值16 |
2.3 at-least-once与 exactly-once语义
用户有三种方式来保存每个分区消费的offset,分别是checkpoint、kafka保存、用户自己处理
- checkpoint:driver端的容错,最不靠谱,由于任务执行与checkpoint是异步执行的,如果消息没有消费完程序报错了,但是已经checkpoint,恢复后则会丢失数据
- kafka保存:设置
enable.auto.commit
为false,需要调用在业务逻辑后加上如下代码。下面的代码注入的offsetRanges
会在下次生成任务的时候被提交。这种方式不会丢数据,当输出端不是幂等的,就只能保证at-least-once,因为输出与保存offset不是一个事务
val kafkaDStream = KafkaUtils.createDirectStream(...)
// 写法一开始
// 一大堆流式处理代码
// 最后
kafkaDStream .foreachRDD { rdd =>
val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offstietRanges
kafkaDStream .asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}
// 写法一结束
// 写法二
kafkaDStream .transform{ rdd =>
val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offstietRanges
kafkaDStream .asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
rdd
}.一大堆流式处理代码
- 用户自己处理,当输出端不幂等,但是支持事务时,将保存offset与输出封装到一个事务中可以实现
exactly-once
,官方文档streaming-kafka-0-10-integration里提供一种写法,笔者在这里也提供一种,供读者参考
// 故障恢复
val fromOffsets = selectOffsetsFromYourDatabase.map { resultSet =>
new TopicPartition(resultSet.string("topic"), resultSet.int("partition")) -> resultSet.long("offset")
}.toMap
val stream = KafkaUtils.createDirectStream(...)
var offsetBroadCast: Broadcast[Array[OffsetRange]] = null
messages.transform(rdd => {
val ranges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
offsetBroadCast = ssc.sparkContext.broadcast(ranges)
rdd
}).map(t => (t.key(), t.value())).foreachRDD(rdd => {
val partitions = rdd.getNumPartitions
val offset = offsetBroadCast.value
rdd.mapPartitionsWithIndex((mapId, it) => Iterator((mapId, it)))
.foreachPartition(it => {
val (mapId, realIt) = it.buffered.head
val commitOffRange = offset(mapid)
// 开启事务
// 用 realIt 更新结果
// 用 commitOffRange 更新对应分区的offset
// 提交事务
})
offsetBroadCast.destroy(true)
})
3. 与receiver-based的差别
这里的receiver指的是ReliableKafkaReceiver
并且开启了WAL,如果不开启WAL则可能会导致数据丢失,存储数据的过程如下:
- 添加到
BlockGenerator
的buffer中- 将buffer中的数据写入到WAL,写入到
BlockManager
- 通知给
ReceiverTracker
,并把元数据写入到WAL
- 更新offset到zookeeper
整个过程能保证数据不丢,但是只能保证at-least-once
语义。差别如下:
- receiver-based需要存储一份数据,kafka里面也有一份,这样的数据冗余没有必要
Direct approach
少了receiver,整个架构更简单- receiver-based的topic partition offset只能由zookeeper管理,
Direct approach
更灵活
以上是关于spark-streaming与kafka的整合的主要内容,如果未能解决你的问题,请参考以下文章
SparkStreaming与Kafka整合遇到的问题及解决方案
Spark-Streaming kafka count 案例
java spark-streaming接收TCP/Kafka数据