聊聊流式批处理

要说流式微批处理，就不得不说一下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端过程。