Spark Streaming+Kakfa细节剖析

本文基于 Kafka 1.1.0 和 Spark 2.3.0 版本源代码进行分析

Kafka 消费接口

Kafka消费接口包含：低级和高级API，这个区分主要针对broker版本 <0.9.0，在0.9.0中引入了新消费者APIorg.apache.kafka.clients.consumer.KafkaConsumer将两者结合起来了。

低级API（<0.9.0）

低级API的入口类为 kafka.consumer.SimpleConsumer，已经被标记为废弃Deprecated，它的替代者是KafkaConsumer。

低级api的好处是可以更多样的控制读取数据，如：

• 同一条消息读取多次
• 只读取一个Topic中的部分Partition
• 管理读取会话，使一条消息只被读取和处理一次（Exactly Once）

使用该API的5个步骤为(引用):

1. Find an active Broker and find out which Broker is the leader for your topic and partition
2. Determine who the replica Brokers are for your topic and partition
3. Build the request defining what data you are interested in
4. Fetch the data
5. Identify and recover from leader changes

高级API（<0.9.0）

更多的时候，我们只关心从kafka中不断的消费新数据，而对上面提到的细节不关心，这些细节包括：find Leader Broker、partition增加减少、consumer增加减少（引起Rebalance）、leader变化、异常处理等。

这时就可以使用高级API了，它的入口是 kafka.consumer.ZookeeperConsumerConnector

要用高级API应该使用多线程开发，每一个线程对应一个Consumer，Consumer和Partition的数量关系为：

• 如果consumer比partition多，是浪费，因为kafka的设计是在一个partition上是不允许并发的，所以consumer数不要大于partition数
• 如果consumer比partition少，一个consumer会对应于多个partitions，这里主要合理分配consumer数和partition数，否则会导致partition里面的数据被取的不均匀
• 如果consumer从多个partition读到数据，不保证数据间的顺序性，kafka只保证在一个partition上数据是有序的，但多个partition，根据你读的顺序会有不同
• 增减consumer，broker，partition会导致rebalance，所以rebalance后consumer对应的partition会发生变化

更详细内容见引用

新消费者API(>=0.9.0)

新API是消费API的重新优化，它利用到kafka内置的组协调协议(group coordination protocol)，它有以下几个优点：

• 重新设计API，结合了低级API和高级API的功能于一体
• 使用pure java开发，且不再依赖于zookeeper和scala了， 老API依赖zookeeper来管理topic的消费组信息
• 新增安全扩展 https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=51809888

详细使用示例见引用

关于Kafka offset的管理

• 提交至ZK，但频繁操作zk是效率比较低的，早期高级消费API offset存在ZK中
• 提交至Kafka集群内部，Kafka 0.9版本后使用高级消费API offset是存在kafka topic: __consumer_offsets
• 自己控制，存储在外部系统中如redis、hdfs等

Spark Streaming + Kafka

spark源码中有两个不同版本的代码：spark-streaming-kafka-0-8 和spark-streaming-kafka-0-10, 对比如下：

image.png

这里以 spark-streaming-kafka-0-8 的进行介绍

Kafka Receiver模式（2.3.0已废弃）

示意图：

image.png

Receiver使用的是高级消费者API，Receiver模式使用代码如下：

import org.apache.spark.streaming.kafka._

val kafkaStream = KafkaUtils.createStream(streamingContext,
    [ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume])

关键点为：

• Topic分区数和Spark RDD分区数是没有关系的，在参数中指定的Topic分区只是增加了，消费者的数量，对应到消费者处理线程数。不会影响spark处理数据的并行度
• 当有多个Topic或者Group时，可以通过创建多个DStream来增加Receiver的数量，从而增加接收数据的并行度，一个Receiver只在一个worker上运行，详细见 level-of-parallelism-in-data-receiving
• 如果启用了WAF，配置为：spark.streaming.receiver.writeAheadLog.enable，则需要设定存储的级别如：StorageLevel.MEMORY_AND_DISK_SER，如果没有启用WAF，Receiver模式不能保证至少一次的语义。（TODO：代码细节）

Receiver模式关联的Spark Conf为spark.streaming.receiver.*.

Direct模式（推荐）

示意图：

image.png

使用新消费API中的低级接口来实现，相比Receiver有以下几个特点：

• 简化并行：不需要通过创建多个DStream并将他们Union的方式，来达到并行读取数据，Direct方式Kafka的Partition和RDD Partition是一致的。
• 更高效：Receiver模式需要开启WAL才可以不丢数据，而Direct没有Receiver，它使用Spark本身的checkout来报错offset（TODO 另外由于Receiver是个单独的Executor异步读取数据，某些情况下内存需求会很大，因此导致整个App的Execotor都需要多申请写内存）
• Exactly-once语义（TODO）

Direct模式没有更新offset到zookeeper中，将导致基于zookeeper的kafka监控工具无法显示数据消费情况，不过这可以通过手动提交offset来实现。
相关配置项为：spark.streaming.kafka.*. 其中关键配置项为spark.streaming.kafka.maxRatePerPartition用于控制最高速率。

Direct模式默认关闭kafka offset自动提交如：

private[kafka010] def fixKafkaParams(kafkaParams: ju.HashMap[String, Object]): Unit = {
logWarning(s"overriding ${ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG} to false for executor")
kafkaParams.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false: java.lang.Boolean)

logWarning(s"overriding ${ConsumerConfig.AUTO_OFFSET_RESET_CONFIG} to none for executor")
kafkaParams.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none")

SparkSteaming+Kafka处理数据Exactly Once介绍

流式处理包含三个部分：

• Receive Data：这里主要依赖于数据源，kafka本身是基于offset的可以保证数据仅读取一次。
• Transformming Data：spark借助rdd的特性保证
• Output Operation：重点要考虑

以下介绍采用Direct模式时保证Exactly Once的三种方式：

• checkpoint+幂等输出操作: 这种方法有缺陷，当代码变更是，checkpoint的数据无法恢复
• Kafka+幂等输出操作：关闭自动提交offset，你需要在确保处理完输出后，再执行提交commitAsync

stream.foreachRDD { rdd =>
val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
// some time later, after outputs have completed
stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}

注：这里类型强转CanCommitOffsets的前提是stream是createDirectStream的结果，没有经过其他转换

• 外部支持事务的存储：将offset和结果数据在一个事务中完成存储

其他问题：

• back pressure的实现方式？
• Spark Checkpoint？

结语

本文只记部分内容，更多内容和细节还是需要看官方文档和代码。

Reference

• https://spark.apache.org/docs/latest/streaming-programming-guide.html
• Kafka Simple Consumer
• streaming-kafka-integration
  • streaming-kafka-0-8-integration
  • streaming-kafka-0-10-integration
• Kafka Consumer API 的使用
• Kafka:High level consumer vs. Low level consumer
• 高级消费API
• Kafka Offset管理方式 https://elang2.github.io/myblog/posts/2017-09-20-Kafak-And-Zookeeper-Offsets.html