Spark Streaming Kafka

Spark Streaming 可以⽀持多种数据源，但目前最常用的还是 Kafka 作为数据源。

Spark Streaming的kafka编程主要包括两种模型

1.基于Receiver模式

2.Direct（无Receiver）模式

 

基于Receiver （spark-streaming-kafka-0-10以上已不支持）

这种方式利用接收器（Receiver）来接收kafka中的数据，其最基本是使用Kafka高阶用户API接口。对于所有的接收器，从kafka接收来的数据会存储在spark的executor中，之后spark streaming提交的job会处理这些数据。

然而，在默认的配置下，这种方式可能会因为底层的失败而丢失数据。如果要启用高可靠机制，让数据零丢失，就必须启用Spark Streaming的预写日志机制（Write Ahead Log，WAL）。该机制会同步地将接收到的Kafka数据写入分布式文件系统（比如HDFS）上的预写日志中。所以，即使底层节点出现了失败，也可以使用预写日志中的数据进行恢复。

需要注意的要点：

1、Kafka中的topic的partition，与Spark中的RDD的partition是没有关系的。所以，在KafkaUtils.createStream()中，提高partition的数量，只会增加一个Receiver中读取partition的线程的数量。不会增加Spark处理数据的并行度。

2、可以创建多个Kafka输入DStream，使用不同的consumer group和topic，来通过多个receiver并行接收数据。

3、如果基于容错的文件系统，比如HDFS，启用了预写日志机制，接收到的数据都会被复制一份到预写日志中。因此，在KafkaUtils.createStream()中，设置的持久化级别是StorageLevel.MEMORY_AND_DISK_SER。

spark streaming启动过后，会选择一台excetor作为ReceiverSupervior

1:Reciver的父级ReciverTracker分发多个job(task)到不同的executor，并启动ReciverSupervisor.

2:ReceiverSupervior会启动对应的实例reciver(kafkareciver,TwitterReceiver),并调用onstart()

3:kafkareciver在通过onstart()启动后就开启线程源源不断的接收数据，并交给ReceiverSupervior，通过ReceiverSupervior.store函数一条一条接收

4:ReceiverSupervior会调用BlockGenertor.adddata填充数据。

所有的中间数据都缓存在BlockGenertor

1:首先BlockGenertor维护了一个缓冲区，currentbuffer，一个无限长度的arraybuffer。为了防止内存撑爆，这个currentbuffer的大小可以被限制，通过设置参数spark.streaming.reciver.maxRate,以秒为单位。currentbuffer所使用的内存不是storage（负责spark计算过程中的所有存储，包括磁盘和内存），而是珍贵的计算内存。所以currentbuffer应该被限制，防止占用过多计算内存，拖慢任务计算效率，甚至有可能拖垮Executor甚至集群。

2:维护blockforpushing队列，它是等待被拉到到BlockManager的中转站。它是currentbuffer和BlockManager的中间环节。它里面的每一个元素其实就是一个currentbuffer。

3:维护两个定时器，其实就是一个生产-消费模式。blockintervaltimer定时器，负责生产端，定时将currentbuffer放进blockforpushing队列。blockforpushingthread负责消费端，定时将blockforpushing里的数据转移到BlockManager。

基于Direct的方式

这种新的不基于Receiver的直接方式，是在Spark 1.3中引入的，从而能够确保更加健壮的机制。替代掉使用Receiver来接收数据后，这种方式会周期性地查询Kafka，来获得每个topic+partition的最新的offset，从而定义每个batch的offset的范围。当处理数据的job启动时，就会使用Kafka的简单consumer api来获取Kafka指定offset范围的数据。它并不需要单独线程实时接收数据，而是每隔batch size时间抓取数据。

这种方式有如下优点：

1、简化并行读取：如果要读取多个partition，不需要创建多个输入DStream然后对它们进行union操作。Spark会创建跟Kafka partition一样多的RDD partition，并且会并行从Kafka中读取数据。所以在Kafka partition和RDD partition之间，有一个一对一的映射关系。

2、高性能：如果要保证零数据丢失，在基于receiver的方式中，需要开启WAL机制。这种方式其实效率低下，因为数据实际上被复制了两份，Kafka自己本身就有高可靠的机制，会对数据复制一份，而这里又会复制一份到WAL中。而基于direct的方式，不依赖Receiver，不需要开启WAL机制，只要Kafka中作了数据的复制，那么就可以通过Kafka的副本进行恢复。

3、一次且仅一次的事务机制：基于receiver的方式，是使用Kafka的高阶API来在ZooKeeper中保存消费过的offset的。这是消费Kafka数据的传统方式。这种方式配合着WAL机制可以保证数据零丢失的高可靠性，但是却无法保证数据被处理一次且仅一次，可能会处理两次。因为Spark和ZooKeeper之间可能是不同步的。

4、降低资源。

Direct不需要Receivers，其申请的Executors全部参与到计算任务中；而Receiver-based则需要专门的Receivers来读取Kafka数据且不参与计算。因此相同的资源申请，Direct 能够支持更大的业务。

5、降低内存。

Receiver-based的Receiver与其他Exectuor是异步的，并持续不断接收数据，对于小业务量的场景还好，如果遇到大业务量时，需要提高Receiver的内存，但是参与计算的Executor并无需那么多的内存。而Direct 因为没有Receiver，而是在计算时读取数据，然后直接计算，所以对内存的要求很低。实际应用中我们可以把原先的10G降至现在的2-4G左右。

6、鲁棒性更好。

Receiver-based方法需要Receivers来异步持续不断的读取数据，因此遇到网络、存储负载等因素，导致实时任务出现堆积，但Receivers却还在持续读取数据，此种情况很容易导致计算崩溃。Direct 则没有这种顾虑，其Driver在触发batch 计算任务时，才会读取数据并计算。队列出现堆积并不会引起程序的失败。

KafkaUtils.createStream接收数据流程

 

实例

以 Kafka 作为数据源的典型基于 Java 的 Spark Streaming 代码框架如下：

public class JavaKafka2Spark { 
public static void main(String args[]) { 
    SparkConf conf = new SparkConf(); 
    JavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.seconds(3)); 
    Map kafkaParams = new HashMap<>(); 
    kafkaParams.put("bootstrap.servers", "host1:9092,host2:9092");        
    kafkaParams.put("key.deserializer", StringDeserializer.class); 
    kafkaParams.put("value.deserializer", StringDeserializer.class);             
    kafkaParams.put("group.id", "test-id"); 
    kafkaParams.put("auto.offset.reset", "latest"); 
    kafkaParams.put("enable.auto.commit", false); 
    Collection topics = Arrays.asList("test"); 
    JavaInputDStream<ConsumerRecord> stream = KafkaUtils.createDirectStream( 
        ssc,     
        LocationStrategies.PreferConsistent(), 
        ConsumerStrategies.Subscribe(topics, kafkaParams) ); 
        final AtomicReference offsetRanges = new AtomicReference<>(); 
        stream.transform(rdd -> 
            {   OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges(); 
                offsetRanges.set(offsets); 
                return rdd; 
            }).map(/*业务逻辑*/).foreachRDD(rdd -> { 
                ((CanCommitOffsets)stream.inputDStream()).commitAsync(offsetRanges.get());
        }); 
    ssc.start(); 
    ssc.awaitTermination(); }
 }

其中，各主要段的含义如下：

1. Spark Streaming 环境初始化

SparkConf conf = new SparkConf();
JavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.seconds(3));

2. 指定 Kafka 连接的参数

Map kafkaParams = new HashMap<>(); kafkaParams.put("bootstrap.servers", "host1:9092,host2:9092"); 
kafkaParams.put("key.deserializer", StringDeserializer.class); kafkaParams.put("value.deserializer", StringDeserializer.class); 
kafkaParams.put("group.id", "test-id"); 
kafkaParams.put("auto.offset.reset", "latest");  // 默认是earliest
kafkaParams.put("enable.auto.commit", false);

3. 指定用于数据读取的 Topic

Collection topics = Arrays.asList("test");

4. 创建 Kafka 连接

JavaInputDStream<ConsumerRecord> stream = KafkaUtils.createDirectStream( 
    ssc, 
    LocationStrategies.PreferConsistent(),
    ConsumerStrategies.Subscribe(topics, kafkaParams) 
);

5. Kafka Offset 处理逻辑：先暂存，再持久化

final AtomicReference offsetRanges = new AtomicReference<>(); 
stream.transform(rdd -> { 
    offsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();           
    offsetRanges.set(offsets); 
    return rdd; }).map(/*业务逻辑*/).foreachRDD(rdd -> 
    { ((CanCommitOffsets) stream.inputDStream()).commitAsync(offsetRanges.get()); });

6. 启动流计算处理进程

ssc.start();

7. 等待，直到用户（或框架）下达终止指令

ssc.awaitTermination();