SDP（0）：Streaming-Data-Processor - Data Processing with Akka-Stream

   再有两天就进入2018了,想想还是要准备一下明年的工作方向。回想当初开始学习函数式编程时的主要目的是想设计一套标准API給那些习惯了OOP方式开发商业应用软件的程序员们，使他们能用一种接近传统数据库软件编程的方式来实现多线程，并行运算，分布式的数据处理应用程序，前提是这种编程方式不需要对函数式编程语言、多线程软件编程以及集群环境下的分布式软件编程方式有很高的经验要求。前面试着发布了一个基于scalaz-stream-fs2的数据处理工具开源项目。该项目基本实现了多线程的数据库数据并行处理，能充分利用域内服务器的多核CPU环境以streaming，non-blocking方式提高数据处理效率。最近刚完成了对整个akka套装（suite）的了解，感觉akka是一套理想的分布式编程工具：一是actor模式提供了多种多线程编程方式，再就是akka-cluster能轻松地实现集群式的分布式编程，而集群环境变化只需要调整配置文件，无需改变代码。akka-stream是一套功能更加完整和强大的streaming工具库，那么如果以akka-stream为基础，设计一套能在集群环境里进行分布式多线程并行数据处理的开源编程工具应该可以是2018的首要任务。同样，用户还是能够按照他们熟悉的数据库应用编程方式轻松实现分布式多线程并行数据处理程序的开发。

   我把一般中小企业的IT系统分成两大部分：一是实时的数据采集（输入）部分，二是批量数据抽取、分析、处理部分。为了让传统中小型企业IT软件编程人员能开发服务器集群环境上数据平台（如云端数据平台）运行的软件系统，我打算通过这个DSP（Streaming-Data-Processor）项目来实现上面提到的第二部分。第一部分可以用CQRS（Command-Query-Responsibility-Separation）即读写分离架构和事件记录（event-sourcing）模式来实现一种高效快速响应、安全稳定运行的数据采集体系。这部分我会在完成SDP项目后以akka-persistence为核心，通过akka-http，AMQP如RabitMQ等技术来实现。

  按一般的scala和akka的编程方式编写多线程分布式数据库管理软件时一是要按照akka代码模式，使用scala编程语言的一些较深的语法；二是需要涉及异步Async调用，集群Cluster节点任务部署及Streaming对外集成actor运算模式的细节，用户需要具备一定的scala，akka使用经验。再接下来就需要按业务流程把各业务环节分解成不依赖顺序的功能模块，然后把这些分拆出来的功能分派给集群中不同的节点上去运算处理。而对于SDP用户来说，具备最基本的scala知识，无需了解akka、actor、threads、cluster，只要按照SDP自定义的业务处理流模式就可以编制多线程分布式数据处理程序了。下面我就用一些文字及伪代码来描述一下SDP的结构和功能：

总体来说SDP是由一或多个Stream组成的；每个Stream就代表一段程序。一段完整的程序Stream是由流元素源Source、处理节点Process-Node（Flow)及数据输出终点Sink三个环节组成，下面是一个典型的程序框架：

  def load(qry: Query): PRG[R,M] = ???
  def process1: PRG[R,M] = ???
  def process2: PRG[R,M] = ???
  def recursiveProcess(prg: PRG[R,M]): PRG[R,M] = ???
  def results: PRG = ???
  
  load(qryOrders).process1.process2.recursiveProcess(subprogram).results.run

从上面的示范中我们可以看到所有定义的函数都产生PRG[R,M]类型结果。其中R类型就是stream的元素，它流动贯穿了程序的所有环节。就像下水道网络运作原理一样：污水由源头Source流入终点Sink，在途中可能经过多个污水处理节点Node。每一个节点代表对管道中流淌污水处理的方式，包括分叉引流、并叉合流、添加化学物质、最后通过终点把处理过的水向外输出。在PRG中流动的R类型可能是数据如数据库表的一行，又或者是一条Sring类型的query如plain-sql，可以用JDBC来运行。cassandra的CQL也是String类型的。Slick，Quill，ScalikeJDBC和一些其它ORM的Query都可以产生plain-sql。

Source是一段程序的开始部分。一般来说Source是通过运算Query产生一串数据行或者人工构建而成。Source也可以并行运算Query产生，然后合并成一条无序的数据源，如下伪代码的类型：

  def load_par(qrys: Query*): PRG[R,M] = ???

Process-Node是SDP最重要的一个组成部分，因为大部分用户定义的各种业务功能是在这里运算的。用户可以选择对业务功能进行拆分然后分派给不同的线程或不同的集群节点进行多线程并行或分布式的运算。SDP应该为用户程序提供多线程，并行式、分布式的运算函数。首先，运算用户程序后应产生R类型结果而且，作为一种reactive软件，必须保证完全消耗上一阶段产生的所有R类型元素。下面是一个用户函数的款式：

  type UserFunc = R => R 

除了fire-and-run类型的运算函数，SDP还应该提供针对多线程或分布式程序的map-reduce式运算函数。初步想法是：无论返回结果与否，分派任务都是由persistence-actor来执行的，这样能保证不会漏掉任何任务。如果整体任务需要在所有分派任务返回运算结果后再统一进行深度运算时akka的actor消息驱动模式是最适合不过的了。具体情况可以参考我前面关于cluster-sharding的博文。

Sink的主要作用实际上是保证完全消耗程序中产生的所有元素，这是reactive类型程序的必须要求。

好了，不知不觉还有几个钟头就进入2017倒计时了。赶快凑合着在跨入2018之前把这篇发布出去，刚好是今年的最后一篇博文。祝各位在新的一年中工作生活称心如意！