spark-streaming与kafka的整合

Posted amddbb

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了spark-streaming与kafka的整合相关的知识,希望对你有一定的参考价值。

1. 概述

在2.x中,spark有两个用来与kafka整合的代码,版本代号为0.80.10,由于在0.8,kafka有两套消费者api,根据高级api得到了Receiver-based Approach,根据低级api得到了Direct Approach,而在0.10由于kafka只有一套消费者api了,所以也只有Direct Approach

2. Direct Approach

由于0.80.10Direct 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个取值

  1. PreferConsistent: 均匀的分布在executor中。原理是使用TopicPartition的hash值来决定到哪个可用的executor上执行
  2. PreferBrokers:偏向于将consumer分配到指定的topic partition所在的leader上;如果leader集中,会造成负载不均。如果可用的executor中没有leader所在的主机,使用 PreferConsistent
  3. PreferFixed: 指定某些topic partition的consumer分配到指定的机器上,没有指定的topic partition使用PreferConsistent

ConsumerStrategy用来指定消费的topic partition,也有三个子类来表示不同的策略:

  1. Subscribe:精确指定用来消费的topic
  2. SubscribePattern:使用正则表达式来指定消费的topic
  3. Assign: 精确指定用来消费的topic partition。上述2个都是消费topic的所有分区,这个可以选择性的消费部分分区

PerPartitionConfig只有一个实现类DefaultPerPartitionConfig,用来执行反压,控制每个分区的消费速率

2.1 执行过程

分为2步,一是DirectKafkaInputDStream计算每个分区及其消费开始的offset与结束的offset(不包含),二是将上述信息封装到KafkaRDD中消费每个topic partition,一个topic partition对应一个task

2.2 常用的配置

配置分为两种,一种是传给ConsumerStrategy,为Kafka Consumer相关的配置,还有一种是传给SparkConf的。
传给ConsumerStrategykafkaParams

变量名描述其它
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,分别是checkpointkafka保存用户自己处理

  1. checkpoint:driver端的容错,最不靠谱,由于任务执行与checkpoint是异步执行的,如果消息没有消费完程序报错了,但是已经checkpoint,恢复后则会丢失数据
  1. 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
}.一大堆流式处理代码
  1. 用户自己处理,当输出端不幂等,但是支持事务时,将保存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则可能会导致数据丢失,存储数据的过程如下:

  1. 添加到BlockGenerator的buffer中
  2. 将buffer中的数据写入到WAL,写入到BlockManager
  3. 通知给ReceiverTracker,并把元数据写入到WAL
  4. 更新offset到zookeeper

整个过程能保证数据不丢,但是只能保证at-least-once语义。差别如下:

  1. receiver-based需要存储一份数据,kafka里面也有一份,这样的数据冗余没有必要
  2. Direct approach少了receiver,整个架构更简单
  3. receiver-based的topic partition offset只能由zookeeper管理,Direct approach更灵活






以上是关于spark-streaming与kafka的整合的主要内容,如果未能解决你的问题,请参考以下文章

SparkStreaming与Kafka整合遇到的问题及解决方案

spark-streaming对接kafka的两种方式

Spark-Streaming kafka count 案例

java spark-streaming接收TCP/Kafka数据

spark-streaming读kafka数据到hive遇到的问题

[Spark]Spark-streaming通过Receiver方式实时消费Kafka流程(Yarn-cluster)