spark streaming 接收kafka消息之一 -- 两种接收方式

Posted 2022-02-12 johnny666888

源码分析的spark版本是1.6。

首先，先看一下 org.apache.spark.streaming.dstream.InputDStream 的 类说明：

This is the abstract base class for all input streams. This class provides methods start() and stop() 
which is called by Spark Streaming system to start and stop receiving data. Input streams that can 
generate RDDs from new data by running a service/thread only on the driver node (that is, without 
running a receiver on worker nodes), can be implemented by directly inheriting this InputDStream. 
For example, FileInputDStream, a subclass of InputDStream, monitors a HDFS directory from the driver 
for new files and generates RDDs with the new files. For implementing input streams that requires 
running a receiver on the worker nodes, use org.apache.spark.streaming.dstream.ReceiverInputDStream 
as the parent class.

翻译如下：

所有输入stream 的抽象父类，这个类提供了 start 和 stop 方法， 这两个方法被spark streaming系统来开始接收或结束接收数据。
两种接收数据的两种方式：
在driver 端接收数据；
1. 输入流通过在driver 节点上运行一个线程或服务，从新数据产生 RDD，继承自 InputDStream 的子类
2. 输入流通过运行在 worker 节点上的一个receiver ，从新数据产生RDD ， 继承自 org.apache.spark.streaming.dstream.ReceiverInputDStream

也就是说 spark 1.6 版本的输入流的抽象父类就是 org.apache.spark.streaming.dstream.InputDStream，其子类如下图所示：

与kafka 对接的两个类已经 在上图中标明。

现在对两种方式做一下简单的比较：

相同点：

1.内部都是通过SimpleConsumer 来获取消息，在获取消息之前，在获取消息之前，from offset 和 until offset 都已经确定。

2.都需要在构造 FetchRequest之前，确定leader， offset 等信息。

3. 其内部都有一个速率评估器，起到平衡速率的作用

不同点：

1. offset 的管理不同。

DirectKafkaInputStream 可以通过外部介质来管理 offset， 比如 redis， mysql等数据库，也可以是hbase等。

KafkaInputStream 则需要使用zookeeper 来管理consumer offset数据， 其内部需要监控zookeeper 的状态。

2. receiver运行的节点不同。

DirectKafkaInputStream 对应的 receiver 是运行在 driver 节点上的。

KafkaInputStream 对应的 receiver 是运行在非driver 的executor 上的。

3. 内部对应的RDD不一样。

DirectKafkaInputStream 对应的是 KafkaRDD，内部的迭代器是KafkaRDDIterator

KafkaInputStream 对应的是 WriteAheadLogBackedBlockRDD 或者是 BlockRDD，内部的迭代器 是自定义的 NextIterator

4. 保证Exactly-once 语义的机制不一样。

DirectKafkaInputStream 是根据 offset 和 KafkaRDD 的机制来保证 exactly-once 语义的

KafkaInputStream 是根据zookeeper的 offset 和WAL 机制来保证 exactly-once 语义的，接收到消息之后，会先保存到checkpoint 的 WAL 中