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个取值

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的。
传给ConsumerStrategy的kafkaParams

注意上述的默认值要根据使用的版本来确定。
传给SparkConf的

2.3 at-least-once与 exactly-once语义

用户有三种方式来保存每个分区消费的offset，分别是checkpoint、kafka保存、用户自己处理

1. checkpoint：driver端的容错，最不靠谱，由于任务执行与checkpoint是异步执行的，如果消息没有消费完程序报错了，但是已经checkpoint，恢复后则会丢失数据

2. 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
}.一大堆流式处理代码

3. 用户自己处理，当输出端不幂等，但是支持事务时，将保存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更灵活