聊聊流式批处理

要说流式微批处理,就不得不说一下TCP流。典型的tcp IO流模型有,bio,伪异步IO,NIO,AIO,Rector模型等。我们这里主要是说伪异步IO。

下面我们一步步将其改造成spark Streaming的 SocketStream。

在伪异步模式,我们是客户端通过TCP链接到服务端。这种在分布式模式下不可行,对于Spark Streaming的微批处理,我们根本不知道Receiver运行在何处,所以,客户端链接都不知道请求到何处,当然,我们也可以做一个复杂的操作来报告我们Receiver的位置。所以,第一步要修改的是将我们的后端改为TCP的client端,然后是client主动链接于外部数据中心也即server端,去拉去或者被push数据。

然后,在上一步改装之后,我们的模型就可能变成如下模式:

也即,client主动去data server建立连接请求,然后开始接收数据,接收数据达到一定的数目,比如1000条(也要有超时机制),然后封装成task扔到线程池中执行。

当然,我们可以对他进行进一步完善,比如,一个线程专门负责接收数据,然后将数据缓存到map或者 Array里,我们在启动一个RecurringTimer也即一个定时线程,每隔一定毫秒,比如200ms,将map或者Array里面的数据封装成一个数据块叫做block,存储于一个内存的Array,然后用一个后台线程阻塞的消费Array中的block并将block存储于一个数据管理器里,比如叫做blockmanager。此时我们再用一个RecurringTimer用来每隔一定时间,如batch=5s,生成一个task,task中有task自身要处理的数据的描述信息,然后放入线程池中去执行,在执行的时候根据数据的描述信息去取0-n个block然后处理。

其实,上述步骤和spark Streaming基于Receiver这种模式很类似。主要区别是,Spark Streaming在执行任务之前必须要先经过spark core的job调度并执行的类似过程,在集群中启动Receiver。由Receiver来完成上述block生成并存储于blockmanager的过程。当然还有就是里面有些线程之间的沟通机制采用的是RPC及EventLoop,具体请参考浪尖的前面的文章。

对于,spark Streaming来说在生成job的时候,实际上在是根据当前批次的数据block信息,封装成了一个叫做BlockRDD的对象,然后就可以根据我们的Spark core的计算方式执行计算操作了,在每个分区生成的task去取对应的block,实际上对于BlockRDD每个block对应与一个partition。

当然了,有些人该问了,spark Streaming不是还可以不急于Receiver么,另一种方式是什么情况呢?

在讨论这个问题之前,我们先谈另一个问题,那就是:有些数据源,比如kafka,数据本身是有分区的概念,而且可以使用offset灵活的获取数据,也即是我们可以通过控制请求偏移,随便去请求我们想要的数据。对于这种数据源,我们完全没必要先把数据取回来存储于blockmanager,然后再从blockmanager里面取出来再去处理(请注意这里先暂时忽略预写日志),这明显很浪费性能。这个时候就出现了一个模式那就是direct模式。也即是我们不用Receiver,生成block,然后构建blockRDD,每个Block当成一个partition;而是在生成job的时候,根据offset信息构建一个叫做KafkaRDD的对象,kafkaRDD里面分区的概念是与kafka内部topic分区一一对应的。然后,再执行spark core的job,计算每个分区生成的task时候,根据KafkaRDD内部的信息去kafka里面具体取数据。

可以看出direct这里面少了,Receiver相关的内容,不需要预写日志,不需要数据来回落地等。提升了很大的性能。

这里只讲了driver端的内部过程,并没有讲Executor端过程。

  • 发表于:
  • 原文链接http://kuaibao.qq.com/s/20180108G0PGNC00?refer=cp_1026
  • 腾讯「云+社区」是腾讯内容开放平台帐号(企鹅号)传播渠道之一,根据《腾讯内容开放平台服务协议》转载发布内容。
  • 如有侵权,请联系 yunjia_community@tencent.com 删除。

同媒体快讯

扫码关注云+社区

领取腾讯云代金券