SparkStreaming消费kafka使用及原理

作者 | 张伟

AI前线出品| ID:ai-front

本文主要介绍Spark Streaming(以下简称SS,版本1.6.3)的一些基本概念,以及SS消费kafka(版本0.8.2.1)数据的两种方式的使用及其原理。我会对这两种方案做详细的解析,同时对比两种方案优劣,以及针对Direct Approach (No Receivers)模式介绍其如何实现Exactly Once Semantics,也就是保证接收到的数据只被处理一次,不丢,不重。

SS是Spark上的一个流式处理框架,可以面向海量数据实现高吞吐量、高容错的实时计算。SS支持多种类型数据源,包括Kafka、Flume、twitter、zeroMQ、Kinesis以及TCP sockets等。SS实时接收数据流,并按照一定的时间间隔(下文称为“批处理时间间隔”)将连续的数据流拆分成一批批离散的数据集;然后应用诸如map、reduce、join和window等丰富的API进行复杂的数据处理;最后提交给Spark引擎进行运算,得到批量结果数据,因此其也被称为准实时处理系统。而结果也能保存在很多地方,如HDFS,数据库等。另外SS也能和MLlib(机器学习)以及GraphX(图计算)完美融合。

《SparkStreaming消费kafka使用及原理》

Spark Streaming支持多种类型数据源

Spark Streaming基础概念

DStream Discretized Stream是SS的基础抽象,代表持续性的数据流和经过各种Spark原语操作后的结果数据流。DStream本质上是一个以时间为键,RDD为值的哈希表,保存了按时间顺序产生的RDD,而每个RDD封装了批处理时间间隔内获取到的数据。SS每次将新产生的RDD添加到哈希表中,而对于已经不再需要的RDD则会从这个哈希表中删除,所以DStream也可以简单地理解为以时间为键的RDD的动态序列。如下图:

《SparkStreaming消费kafka使用及原理》

窗口时间间隔

窗口时间间隔又称为窗口长度,它是一个抽象的时间概念,决定了SS对RDD序列进行处理的范围与粒度,即用户可以通过设置窗口长度来对一定时间范围内的数据进行统计和分析。假如设置批处理时间间隔为1s,窗口时间间隔为3s。如下图,DStream每1s会产生一个RDD,红色边框的矩形框就表示窗口时间间隔,一个窗口时间间隔内最多有3个RDD,Spark Streaming在一个窗口时间间隔内最多会对3个RDD中的数据进行统计和分析。

《SparkStreaming消费kafka使用及原理》

滑动时间间隔

滑动时间间隔决定了SS程序对数据进行统计和分析的频率。它指的是经过多长时间窗口滑动一次形成新的窗口,滑动时间间隔默认情况下和批处理时间间隔相同,而窗口时间间隔一般设置的要比它们两个大。在这里必须注意的一点是滑动时间间隔和窗口时间间隔的大小一定得设置为批处理时间间隔的整数倍。

如下图,批处理时间间隔是1个时间单位,窗口时间间隔是3个时间单位,滑动时间间隔是2个时间单位。对于初始的窗口time 1-time 3,只有窗口时间间隔满足了才触发数据的处理。这里需要注意的一点是,初始的窗口有可能覆盖的数据没有3个时间单位,但是随着时间的推进,窗口最终会覆盖到3个时间单位的数据。当每个2个时间单位,窗口滑动一次后,会有新的数据流入窗口,这时窗口会移去最早的两个时间单位的数据,而与最新的两个时间单位的数据进行汇总形成新的窗口(time3-time5)。

《SparkStreaming消费kafka使用及原理》

Spark Streaming读取kafka数据

Spark Streaming 与Kafka集成接收数据的方式有两种:

1. Receiver-based Approach

2. Direct Approach (No Receivers)

Receiver-based Approach

这个方法使用了Receivers来接收数据。Receivers的实现使用到Kafka高级消费者API。对于所有的Receivers,接收到的数据将会保存在Spark executors中,然后由SS启动的Job来处理这些数据。然而,在默认的配置下,这种方法在失败的情况下会丢失数据,为了保证零数据丢失,你可以在SS中使用WAL日志,这是在Spark  1.2.0才引入的功能,这使得我们可以将接收到的数据保存到WAL中(WAL日志可以存储在HDFS上),所以在失败的时候,我们可以从WAL中恢复,而不至于丢失数据。架构图如下:

《SparkStreaming消费kafka使用及原理》

使用方式:

(1) 导入kafka的Spark Streaming整合包

《SparkStreaming消费kafka使用及原理》

(2) 创建DStream

《SparkStreaming消费kafka使用及原理》

《SparkStreaming消费kafka使用及原理》

《SparkStreaming消费kafka使用及原理》

需要注意的几点:

1) kafka的topic和partition并不和SS生成的RDD的partition相对应,所以上面代码中topicMap里增加threads只能增加使用一个receiver消费这个topic的线程数,它并不能增加Spark处理数据的并行数,因为每个input DStream在一个worker机器上只创建一个接受单个数据流的receiver。

2) 可以为不同topic和group创建多个DStream来使用多个receiver并行的接受数据。例如:一个单独的kafka input DStream接受两个topic的数据可以分为两个kafka input DStream,每个只接受一个topic的数据,这样可以并行的接受速度从而提高整体吞吐量。

3) 如果开启了wal来保证数据不丢失话,需要设置checkpoint目录,并且像上面代码一样指定数据序列化到hdfs上的方式(比如:StorageLevel.MEMORY_AND_DISK_SER)

4) 建议每个批处理时间间隔周期接受到的数据最好不要超过接受Executor的内存(Storage)的一半。要描述清楚 Receiver-based Approach ,我们需要了解其接收流程,分析其内存使用,以及相关参数配置对内存的影响。

数据接收流程

当执行SS的start方法后,SS会标记StreamingContext为Active状态,并且单独起个线程通过ReceiverTracker将从ReceiverInputDStreams中获取的receivers以并行集合的方式分发到worker节点,并运行他们。worker节点会启动ReceiverSupervisor。接着按如下步骤处理:

(1) ReceiverSupervisor会启动对应的Receiver(这里是KafkaReceiver)

(2) KafkaReceiver 会根据配置启动新的线程接受数据,在该线程中调用                            ReceiverSupervisor.pushSingle方法填充数据,注意,这里是一条一条填充的。

3) ReceiverSupervisor 会调用 BlockGenerator.addData 进行数据填充。

到目前为止,整个过程不会有太多内存消耗,正常的一个线性调用。所有复杂的数据结构都隐含在BlockGenerator中。

BlockGenerator 存储结构

BlockGenerator 会复杂些,重要的数据存储结构有四个:

1) 维护了一个缓存 currentBuffer ,这是一个变长的数组的ArrayBuffer。currentBuffer并不会被复用,而是每个spark.streaming.blockInterval都会新建一个空的变长数据替换老的数据作为新的currentBuffer,然后把老的对象直接封装成Block放入到blocksForPushing的队列里,BlockGenerator会负责保证currentBuffer 只有一个。currentBuffer填充的速度是可以被限制的,以秒为单位,配置参数为spark.streaming.receiver.maxRate,是单个Receiver 每秒钟允许添加的条数。这个是Spark内存控制的第一步,填充currentBuffer 是阻塞的,消费Kafka的线程直接做填充。

2) 维护了一个 blocksForPushing的阻塞队列,size 默认为10个(1.6.3版本),可通过spark.streaming.blockQueueSize进行配置。该队列主要用来实现生产-消费模式,每个元素其实是一个currentBuffer形成的block。

3) blockIntervalTimer 是一个定时器。其实是一个生产者,负责将当前currentBuffer的数据放到blocksForPushing 中,并新建一个currentBuffer。通过参数spark.streaming.blockInterval 设置,默认为200ms。放的方式很简单,直接把currentBuffer做为Block的数据源。这就是为什么currentBuffer不会被复用。

4) blockPushingThread 也是一个定时器,负责将Block从blocksForPushing取出来,然后交给BlockManagerBasedBlockHandler.storeBlock方法。10毫秒会取一次,不可配置。到这一步,才真的将数据放到了Spark的BlockManager中。

下面我们会详细分析每一个存储对象对内存的使用情况:

currentBuffer

首先自然要说下currentBuffer,它缓存的数据会被定时器每隔spark.streaming.blockInterval

(默认200ms)的时间拿走,这个缓存用的是Spark的运行时内存(我们使用的是静态内存管理模式,默认应该是heap*0.2,如果是统一内存管理模式的话应该是heap*0.25),而不是storage内存。如果200ms期间你从Kafka接受的数据足够大,则这部分内存很容易OOM或者进行大量的GC,导致receiver所在的Executor极容易挂掉或者处理速度也很慢。如果你在SparkUI发现Receiver挂掉了,考虑有没有可能是这个问题。

blocksForPushing

blocksForPushing这个是作为currentBuffer 和BlockManager之间的中转站。默认存储的数据最大可以达到 10*currentBuffer 大小。一般不大可能有问题,除非你的spark.streaming.blockInterval设置的比10ms 还小,官方推荐最小也要设置成 50ms,只要你不设置的过大,这块不用太担心。

blockPushingThread

blockPushingThread负责从 blocksForPushing 获取数据,并且写入BlockManager。blockPushingThread只写他自己所在的Executor的 blockManager,也就是一个receiver每个批处理时间间隔周期的数据都会被一个Executor接收。这是导致内存被撑爆的最大风险,在数据量很大的情况下,会导致Receiver所在的Executor直接挂掉。 对应的解决方案在上面需要注意的建议4)有提到,也可以使用多个Receiver来消费同一个topic,降低每个receiver接收的数据量,使用类似下面的代码

《SparkStreaming消费kafka使用及原理》

《SparkStreaming消费kafka使用及原理》

《SparkStreaming消费kafka使用及原理》

前面我们提到,SS的消费速度可以设置上限,其实SS也可以根据之前的周期处理情况来自动调整下一个周期处理的数据量。你可以通过将 spark.streaming.backpressure.enabled 设置为true打开该功能。算法的论文可参考: Socc 2014:AdaptiveStream Processing using Dynamic Batch Sizing,还是有用的,我现在也都开启着。

另外,Spark里除了这个Dynamic,还有一个就是Dynamic Allocation,也就是Executor数量会根据资源使用情况,自动分配资源。具体见官网文档

Direct Approach (No Receivers)

和基于Receiver接收数据不一样,这种方式定期地从Kafka的topic+partition中查询最新的偏移量,再根据定义的偏移量范围在每个批处理时间间隔里面处理数据。当作业需要处理的数据来临时,Spark通过调用Kafka的低级消费者API读取一定范围的数据。这个特性目前还处于试验阶段,而且仅仅在Scala和Java语言中提供相应的API。

和基于Receiver方式相比,这种方式主要有一些几个优点:

(1)简化并行。我们不需要创建多个Kafka输入流,然后union他们。而使用DirectStream,SS将会创建和Kafka分区一样的RDD分区个数,而且会从Kafka并行地读取数据,也就是说Spark分区将会和Kafka分区有一一对应的关系,这对我们来说很容易理解和使用;

(2)高效。第一种实现零数据丢失是通过将数据预先保存在WAL中,这将会复制一遍数据,这种方式实际上很不高效,因为这导致了数据被拷贝两次:一次是被Kafka复制;另一次是写到WAL中。但是本方法因为没有Receiver,从而消除了这个问题,所以不需要WAL日志;

(3)恰好一次语义(Exactly-once semantics)。第一种实现中通过使用Kafka高层次的API把偏移量写入Zookeeper中,这是读取Kafka中数据的传统方法。虽然这种方法可以保证零数据丢失,但是还是存在一些情况导致数据会丢失,因为在失败情况下通过SS读取偏移量和Zookeeper中存储的偏移量可能不一致。而本文提到的方法是通过Kafka低层次的API,并没有使用到Zookeeper,偏移量仅仅被SS保存在Checkpoint中。这就消除了SS和Zookeeper中偏移量的不一致,而且可以保证每个记录仅仅被SS读取一次,即使是出现故障。但是本方法唯一的坏处就是没有更新Zookeeper中的偏移量,所以基于Zookeeper的Kafka监控工具将会无法显示消费的状况。但是你可以通过自己手动地将偏移量写入到Zookeeper中。

架构图如下:

《SparkStreaming消费kafka使用及原理》

使用方式:

《SparkStreaming消费kafka使用及原理》

个人认为,DirectApproach更符合Spark的思维。我们知道,RDD的概念是一个不变的,分区的数据集合。我们将kafka数据源包裹成了一个KafkaRDD,RDD里的partition 对应的数据源为kafka的partition。唯一的区别是数据在Kafka里而不是事先被放到Spark内存里。其实包括FileInputStream里也是把每个文件映射成一个RDD,比较好奇,为什么一开始会有Receiver-based Approach,额外添加了Receiver这么一个概念。

DirectKafkaInputDStream

SS通过Direct Approach接收数据的入口自然是KafkaUtils.createDirectStream 了。在调用该方法时,会先创建

            protected valkc=newKafkaCluster(kafkaParams)

KafkaCluster这个类是真实负责和Kafka 交互的类,该类会获取Kafka的partition信息,接着会创建DirectKafkaInputDStream。此时会获取每个Topic的每个partition的offset。 如果配置成smallest则拿到最早的offset,否则拿最近的offset。

每个DirectKafkaInputDStream 也会持有一个KafkaCluster实例。

到了计算周期后,对应的DirectKafkaInputDStream .compute方法会被调用,此时做下面几个操作:

1) 获取对应Kafka Partition的untilOffset。这样就确定了需要获取数据的offset的范围,同时也就知道了需要计算多少数据了

2) 构建一个KafkaRDD实例。这里我们可以看到,每个计算周期里,DirectKafkaInputDStream和 KafkaRDD 是一一对应的

3) 将相关的offset信息报给InputInfoTracker

4) 返回该RDD

KafkaRDD 的组成结构

KafkaRDD

包含 N(N=Kafka的partition数目)个 KafkaRDDPartition,每个KafkaRDDPartition 其实只是包含一些信息,譬如topic,offset等,真正如果想要拉数据,是通过KafkaRDDIterator 来完成,一个KafkaRDDIterator对应一个 KafkaRDDPartition。整个过程都是延时过程,也就是说数据其实都还在Kafka里,直到有实际的action被触发,才会主动去kafka拉数据。

限速

Direct Approach ( NoReceivers ) 的接收方式也是可以限制接受数据的量的。你可以通过设置 spark.streaming.kafka.maxRatePerPartition来完成对应的配置。需要注意的是,这里是对每个Partition进行限速。所以你需要事先知道Kafka有多少个分区,才好评估系统的实际吞吐量,从而设置该值。

相应的,spark.streaming.backpressure.enabled参数在Direct Approach 中也是继续有效的。

Receiver-based Approach VS Direct Approach (No Receivers)

经过上面对两种数据接收方案的介绍,我们发现, Receiver-based Approach 存在各种内存折腾,对应的Direct Approach (No Receivers)则显得比较纯粹简单些,这也给其带来了较多的优势,主要有如下几点:

1) 因为按需要拉数据,所以不存在缓冲区,就不用担心缓冲区把内存撑爆了。这个在Receiver-based Approach就比较麻烦,你需要通过spark.streaming.blockInterval等参数来调整。

2) 数据默认就被分布到了多个Executor上。Receiver-based Approach 你需要做特定的处理,才能让Receiver分不到多个Executor上。

3) Receiver-based Approach 的方式,一旦你的Batch Processing 被delay了,或者被delay了很多个batch,那估计你的Spark Streaming程序离崩溃也就不远了。 Direct Approach (No Receivers) 则完全不会存在类似问题。就算你delay了很多个batch time,你内存中的数据只有这次处理的。

4) Direct Approach (No Receivers) 直接维护了 Kafka offset,可以保证数据只有被执行成功了,才会被记录下来,通过checkpoint机制。如果采用Receiver-based Approach,消费Kafka和数据处理是被分开的,这样就很不好做容错机制,比如系统宕掉了。所以你需要开启WAL,但是开启WAL带来一个问题是,数据量很大,对HDFS是个很大的负担,而且也会给实时程序带来比较大延迟。

我原先以为Direct Approach 因为只有在计算的时候才拉取数据,可能会比Receiver-based Approach 的方式慢,但是经过我自己的实际测试,总体性能 Direct Approach会更快些,因为Receiver-based Approach可能会有较大的内存隐患,GC也会影响整体处理速度。

如何保证数据接收的可靠性

SS 自身可以做到 at least once 语义,具体方式是通过CheckPoint机制。

CheckPoint 机制

CheckPoint 会涉及到一些类,以及他们之间的关系:

DStreamGraph类负责生成任务执行图,而JobGenerator则是任务真实的提交者。任务的数据源则来源于DirectKafkaInputDStream,checkPoint 一些相关信息则是由类DirectKafkaInputDStreamCheckpointData负责。好像涉及的类有点多,其实没关系,我们完全可以不用关心他们。先看看checkpoint都干了些啥,checkpoint 其实就序列化了一个类而已:org.apache.spark.streaming.Checkpoint

以下是其中的类成员:

《SparkStreaming消费kafka使用及原理》

其他的都比较容易理解,最重要的是 graph,该类全路径名是:                                                              org.apache.spark.streaming.DStreamGraph

里面有两个核心的数据结构是:

private valinputStreams=newArrayBuffer[InputDStream[_]]()private valoutputStreams=newArrayBuffer[DStream[_]]()

inputStreams 对应的就是DirectKafkaInputDStream 了。

再进一步,DirectKafkaInputDStream 有一个重要的对象

protected[streaming]override valcheckpointData=newDirectKafkaInputDStreamCheckpointData

checkpointData 里则有一个data 对象,里面存储的内容也很简单data.asInstanceOf[mutable.HashMap[Time,Array[OffsetRange.OffsetRangeTuple]]]

    就是每个batch 的唯一标识time 对象,以及每个KafkaRDD对应的的Kafka偏移信息。

而 outputStreams 里则是RDD,如果你存储的时候做了foreach操作,那么应该就是forEachRDD了,他被序列化的时候是不包含数据的。

经过上面的分析,我们发现:

1) checkpoint 是非常高效的。没有涉及到实际数据的存储。一般大小只有几十K,因为只存了Kafka的偏移量等信息。

2) checkpoint 采用的是序列化机制,尤其是DStreamGraph的引入,里面包含了可能如ForeachRDD等,而ForeachRDD里面的函数应该也会被序列化。如果采用了CheckPoint机制,而你的程序包做了做了变更,恢复后可能会有一定的问题(这个在测试过程中碰到过)。接着我们看看JobGenerator是怎么提交一个真实的batch任务的,分析在什么时间做checkpoint 操作,从而保证数据的高可用:

1) 产生jobs

2) 成功则提交jobs 然后异步执行

3) 失败则会发出一个失败的事件

4) 无论成功或者失败,都会发出一个 DoCheckpoint 事件。

5) 当任务运行完成后,还会再调用一次DoCheckpoint 事件。

只要任务运行完成后没能顺利执行完DoCheckpoint前crash,都会导致这次Batch被重新调度。也就说无论怎样,不存在丢数据的问题,而这种稳定性是靠checkpoint机制以及Kafka的可回溯性来完成的。

那现在会产生一个问题,假设我们的业务逻辑会对每一条数据都处理,则

1) 我们没有处理一条数据

2) 我们可能只处理了部分数据

3) 我们处理了全部数据

根据我们上面的分析,无论如何,这次失败了,都会被重新调度,那么我们可能会重复处理数据。有可能事最后失败的那一批次数据的一部分,也可能是全部,但不会更多了。

业务需要做事务,保证 Exactly Once 语义

这里业务场景被区分为两个:

1) 幂等操作

2) 业务代码需要自身添加事物操作

所谓幂等操作就是重复执行不会产生问题,如果是这种场景下,你不需要额外做任何工作。但如果你的应用场景是不允许数据被重复执行的,那只能通过业务自身的逻辑代码来解决了。

这个SS 倒是也给出了官方方案:

《SparkStreaming消费kafka使用及原理》

这代码什么含义呢?就是说针对每个partition的数据,产生一个uniqueId,只有这个partion的所有数据被完全消费,则算成功,否则算失败,要回滚。下次重复执行这个uniqueId时,如果已经被执行成功过的,则skip掉。这样,就能保证数据Exactly Once 语义了。

总结

根据我的实际经验,目前Direct Approach 稳定性个人感觉比 Receiver-based Approach 更好些,推荐使用 Direct Approach 方式和Kafka进行集成,并且开启相应的checkpoint 功能,保证数据接收的稳定性,Direct Approach 模式本身可以保证数据 at least once语义,如果你需要Exactly Once 语义时,需要保证你的业务是幂等,或者保证了相应的事务。

作者介绍

张伟,TalkingData数据工程师,5年软件开发经验,3年大数据相关工作经验,擅长离线计算、批计算、nosql数据库等。

-全文完-

关注人工智能的落地实践,与企业一起探寻 AI 的边界,AICon 全球人工智能技术大会火热售票中,8 折倒计时一周抢票,详情点击:

http://t.cn/Rl2MGtT

《深入浅出TensorFlow》迷你书现已发布,关注公众号“AI前线”,ID:ai-front,回复关键字:TF,获取下载链接!

《SparkStreaming消费kafka使用及原理》

    原文作者:AI前线
    原文地址: https://www.jianshu.com/p/200f6e6fca48
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞