sparkstreaming读取kafka的两种方式

spark streaming提供了两种获取方式，一种是同storm一样，实时读取缓存到内存中;另一种是定时批量读取。

这两种方式分别是：

Receiver-base

Direct

一、Receiver-base：
Spark官方最先提供了基于Receiver的Kafka数据消费模式。不过这种方式是先把数据从kafka中读取出来，然后缓存在内存，再定时处理。如果这时候集群退出，而偏移量又没处理好的话，数据就丢掉了，存在程序失败丢失数据的可能，后在Spark 1.2时引入一个配置参数spark.streaming.receiver.writeAheadLog.enable以规避此风险。
Receiver-based的Kafka读取方式是基于Kafka高阶(high-level) api来实现对Kafka数据的消费。在提交Spark Streaming任务后，Spark集群会划出指定的Receivers来专门、持续不断、异步读取Kafka的数据，读取时间间隔以及每次读取offsets范围可以由参数来配置。读取的数据保存在Receiver中，具体StorageLevel方式由用户指定，诸如MEMORY_ONLY等。当driver 触发batch任务的时候，Receivers中的数据会转移到剩余的Executors中去执行。在执行完之后，Receivers会相应更新ZooKeeper的offsets。如要确保at least once的读取方式，可以设置spark.streaming.receiver.writeAheadLog.enable为true。具体Receiver执行流程见下图：
sparkstreaming读取kafka的两种方式
Spark Streaming提供了一些重载读取Kafka数据的方法，本文中关注两个基于Scala的方法，这在我们的应用场景中会用到，具体的方法代码如下：

方法createDirectStream中，ssc是StreamingContext；kafkaParams的具体配置见Receiver-based之中的配置，与之一样；这里面需要指出的是fromOffsets ，其用来指定从什么offset处开始读取数据。
sparkstreaming读取kafka的两种方式
方法createDirectStream中，该方法只需要3个参数，其中kafkaParams还是一样，并未有什么变化，不过其中有个配置auto.offset.reset可以用来指定是从largest或者是smallest处开始读取数据；topic是指Kafka中的topic，可以指定多个。具体提供的方法代码如下：
sparkstreaming读取kafka的两种方式
在实际的应用场景中，我们会将两种方法结合起来使用，大体的方向分为两个方面：

应用启动。当程序开发并上线，还未消费Kafka数据，此时从largest处读取数据，采用第二种方法；

应用重启。因资源、网络等其他原因导致程序失败重启时，需要保证从上次的offsets处开始读取数据，此时就需要采用第一种方法来保证我们的场景

总体方向上，我们采用以上方法满足我们的需要，当然具体的策略我们不在本篇中讨论，后续会有专门的文章来介绍。从largest或者是smallest处读Kafka数据代码实现如下：
sparkstreaming读取kafka的两种方式
程序失败重启的逻辑代码如下：

代码中的fromOffsets参数从外部存储获取并需要处理转换，其代码如下：

该方法提供了从指定offsets处读取Kafka数据。如果发现读取数据异常，我们认为是offsets失败，此种情况去捕获这个异常，然后从largest处读取Kafka数据。

Receive_base VS Direct两种方式的优缺点：
Direct方式具有以下方面的优势：
1、简化并行(Simplified Parallelism)。不现需要创建以及union多输入源，Kafka topic的partition与RDD的partition一一对应
2、高效(Efficiency)。Receiver-based保证数据零丢失(zero-data loss)需要配置spark.streaming.receiver.writeAheadLog.enable,此种方式需要保存两份数据，浪费存储空间也影响效率。而Direct方式则不存在这个问题。
3、强一致语义(Exactly-once semantics)。High-level数据由Spark Streaming消费，但是Offsets则是由Zookeeper保存。通过参数配置，可以实现at-least once消费，此种情况有重复消费数据的可能。
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 计算任务时，才会读取数据并计算。队列出现堆积并不会引起程序的失败。

Direct方式的缺点：
提高成本。Direct需要用户采用checkpoint或者第三方存储来维护offsets，而不像Receiver-based那样，通过ZooKeeper来维护Offsets，此提高了用户的开发成本。

监控可视化。Receiver-based方式指定topic指定consumer的消费情况均能通过ZooKeeper来监控，而Direct则没有这种便利，如果做到监控并可视化，则需要投入人力开发。

Receive-base优点：
1、Kafka的high-level数据读取方式让用户可以专注于所读数据，而不用关注或维护consumer的offsets，这减少用户的工作量以及代码量而且相对比较简单。

Receive-base的缺点：
1、防数据丢失。做checkpoint操作以及配置spark.streaming.receiver.writeAheadLog.enable参数，配置spark.streaming.receiver.writeAheadLog.enable参数，每次处理之前需要将该batch内的日志备份到checkpoint目录中，这降低了数据处理效率，反过来又加重了Receiver端的压力；另外由于数据备份机制，会受到负载影响，负载一高就会出现延迟的风险，导致应用崩溃。
2、单Receiver内存。由于receiver也是属于Executor的一部分，那么为了提高吞吐量，提高Receiver的内存。但是在每次batch计算中，参与计算的batch并不会使用到这么多的内存，导致资源严重浪费。
3、在程序失败恢复时，有可能出现数据部分落地，但是程序失败，未更新offsets的情况，这导致数据重复消费。
4、提高并行度，采用多个Receiver来保存Kafka的数据。Receiver读取数据是异步的，并不参与计算。如果开较高的并行度来平衡吞吐量很不划算。5、Receiver和计算的Executor的异步的，那么遇到网络等因素原因，导致计算出现延迟，计算队列一直在增加，而Receiver则在一直接收数据，这非常容易导致程序崩溃。
6、采用MEMORY_AND_DISK_SER降低对内存的要求。但是在一定程度上影响计算的速度

sparkstreaming读取kafka的两种方式

相关推荐