S4分布式流计算引擎

背景

  最近花了点时间研究了下分布式计算这一块的内容。领导给的第一个任务，就是学习下S4和GridGain。花了几天的时间把s4的源码看了下，把自己的理解和学习的内容做一个记录。 下一篇会是GridGain的分享

学习

• s4的论文： S4: Distributed Stream Computing Platform
• s4的官网： http://s4.io/

s4是什么？

1.  s4的全称 ：  Simple Scalable Streaming System (简单的描述：分布式流计算系统)

2.  特点： 

• distributed(分布式)
• scalable(扩展性)
• partially fault-toleran(部分容错性)
• pluggable (可插拔)

3. 产生的原因： 

• Yahoo发起，主要用于解决"cost-per-click“广告，通过实时计算预测用户对广告的可能的点击行为。
• 不用hadoop的原因： hadoop主要解决batch处理，基于mapReduce对可控的数据的进行处理。而流计算是针对不可控的点击事件，对实时性有严格要求。

4.  适用的场景：

• 业务允许部分容错性。 (s4没有严格的failover机制，运行节点突然crash时，会导致当前节点中的数据丢失。后续的请求会failover到其他的节点上)

 

S4的设计：

 

容器概念(http://docs.s4.io/manual/overview.html)

1. PE : Processing Elements (处理节点)
   * emit one or more events which may be consumed by other PEs,
   * publish results, possibly to an external data store or consumer.
2. Events :  message   (消息)
   * arbitrary Java Objects
   * passed between PEs. (send and receive)
3. PEC : processing element container  (处理节点容器)
   * invokes the appropriate PEs in the appropriate order
4. node : a Processing Endpoint  (机器节点)
   * a jvm instance
   * contains a PEC
5. cluster: a group nodes  (节点集群)

说明：

• 一个cluster包含多个node
• 每个node包含一个PE Container
• 每个PE Container 包含多个PE
• 每个PE消费Events，生成新的Events并传递给下一个PE

 

总体结构图：

1. PE Container/PE
2. EventListenter
3. Dispatcher, EventEmitter
4. Communcation

 

 

 

 

PE内部概念：(4个部分组成)

• its functionality as defined by a PE class and associated configuration,
• the named stream that it consumes,
• the keyed attribute in those events, and
• the value of the keyed attribute in events which it consumers

PE类关系图：

说明：

• Persister　：　每个PE对应一个Persister，s4中每个PE对应于一个key的value结果。每个value结构都将作为PE的一个instance
• FrequenceType :  每个PE会定期进行flush output输出，可选择的策略(定时，数量阀值)
• Clock ： 每个PE的时间控制单元，有几种时间。(WallClock:基于系统时间处理 , EventClock：基于event事件时间控制)

重点理解一下: Keyless PE概念和PE Prototype

• PE在底层实现了会以多实例存在。存储的key即为其keyed对应的value值，内部有个lookupTable概念。
• 针对Keyless PE，其对应的存储key即为"*"，所以每次通过lookupTable.get(value)返回的即为同一个节点，单例化
• 针对prototype，其对应的存储就为其value，所以每次会根据当前keyed attribute确定返回的PE节点，基于这点可以实现PE节点数据的Join处理

 

EventListener/EventEmitter：

 

 说明：　

• 每个PE Container是一个EventProducer，使用CommLayerListener做为其事件处理器，处理EventWrapper反序列化。
• 每个PE包含一个Dispatcher，Dispatcher里包含了一个EventEmitter处理对应的EventWrapper对象的发送
• 底层实际通讯的类：　SenderProcess/ListenerProcess

 

Dispatcher类关系图：

 

说明：　

• Partitioner，　每个dispatcher针对发送的目标cluster，会根据对应的key进行分区处理，路由到其中的node节点。(node节点的信息可以通过zookeeper进行动态管理)

 

考虑集群node节点的管理(node的新增 or 修改)

说明：

• ProcessMonitor 监控当前运行node节点的状态，有static/dymaic两种维护状态
• TaskManager  创建node节点，主要是设置lock文件，有static/dymaic两种维护状态

运行(PE状态变化)

 

S4缺点：

S4产品还是一个半成品，整体代码结构组织和风格上还是比较乱的，选择使用时需谨慎。存在的一些问题：

1. failover (运行node节点出现crash，当前node上的PE数据将无法实施failover)
2. persist (目前支持方式过于简单，需要考虑网络持久化，类似于nfs，分布式文件系统等，配合failover机制)
3. communication  (只支持udp协议，数据传输可靠性上)
4. load balancer (根据系统负载进行智能LB，目前暂时未看到相关实现。系统运行分为两种模式static or dymaic模式, static不存在智能调节LB处理)
5. deploy  (手工方式介入deploy，无法支持apps的zero deploy模式。系统分为cluster/node两概念，node对应于一计算节点实例,cluster为一组处理相同业务的计算节点)

评论

6 楼 AliKevin2011 2011-10-14  

我目前看到s4的整体内部处理是下面的过程，不知道是否正确？
1.client发送数据到adapter的clientSub.
2.adapter.clientSub对数据进行Wrapper和injectEvent
3.adapter.EventDispatcher分配事件到s4-CommLayerListener-ListenerProcess-GenericListener
4.s4-CommLayerListener-ListenerProcess.listen()-messageQueue
5.s4-CommLayerListener.Dequeuer.takeMessage()
6.s4-CommLayerListener.handler.processEvent-PEContainer.queueWork.workQueue
7.s4-CommLayerListenerhandle.PEContainer.CheckpointingOrRecovery
8.PE do business.
9.finish
望指点。我有些模糊。

5 楼 lianlupengUestc 2011-10-03  

厉害，最近准备研究一把呢，嘿嘿，麻烦把您的邮箱给我吧，以后有问题找你啊

4 楼 agapple 2011-06-22  

lance_123 写道

agapple 写道

lance_123 写道

failover是非常致命的地方。
之前本来打算拿来项目用，看了一阵子后，发现非常致命，特别是在有些业务下。
不过框架思想很值得学习。

看来是taobao的同学。 s4从它产生的项目来看，主要就是解决实时计算的需求，允许请求丢失。但在一些商业项目中，是不会允许出现数据丢失。

s4针对PE节点数据，有个Persist概念，想做failover也靠谱，借助于类似共享存储，性能上估计会有比较大的损失。

做persist的话，性能和架构变得复杂。我现在自已在写一些这方面的demo。
期待你下次的GridGain分享。

客气了，看了下你在csdn上的blog，在分布式方面,hadoop的分析上还是蛮有料的。
最近在忙着做BeanMapping开源工具功能完善的事，分布式的学习计划只能在工作内容的30%，进度上已经拉下一截。

3 楼 lance_123 2011-06-22  

agapple 写道

lance_123 写道

failover是非常致命的地方。
之前本来打算拿来项目用，看了一阵子后，发现非常致命，特别是在有些业务下。
不过框架思想很值得学习。

看来是taobao的同学。 s4从它产生的项目来看，主要就是解决实时计算的需求，允许请求丢失。但在一些商业项目中，是不会允许出现数据丢失。

s4针对PE节点数据，有个Persist概念，想做failover也靠谱，借助于类似共享存储，性能上估计会有比较大的损失。

做persist的话，性能和架构变得复杂。我现在自已在写一些这方面的demo。
期待你下次的GridGain分享。

2 楼 agapple 2011-06-22  

lance_123 写道

failover是非常致命的地方。
之前本来打算拿来项目用，看了一阵子后，发现非常致命，特别是在有些业务下。
不过框架思想很值得学习。

看来是taobao的同学。 s4从它产生的项目来看，主要就是解决实时计算的需求，允许请求丢失。但在一些商业项目中，是不会允许出现数据丢失。

s4针对PE节点数据，有个Persist概念，想做failover也靠谱，借助于类似共享存储，性能上估计会有比较大的损失。

1 楼 lance_123 2011-06-21  

failover是非常致命的地方。
之前本来打算拿来项目用，看了一阵子后，发现非常致命，特别是在有些业务下。
不过框架思想很值得学习。