《大规模分布式存储系统》第13章 大数据【13.4】

javazx

13.4　流式计算
MapReduce及其扩展解决了离线批处理问题，但是无法保证实时性。对于实时性
要求高的场景，可以采用流式计算或者实时分析系统进行处理。
7 f; P4 t! r. u: u5 b0 v* r, K流式计算（Stream Processing）解决在线聚合（Online Aggregation）、在线过滤
（Online Filter）等问题，流式计算同时具有存储系统和计算系统的特点，经常应用
. I5 E0 ^6 z1 x. {9 k1 Y在一些类似反作弊、交易异常监控等场景。流式计算的操作算子和时间相关，处理
8 H  w6 |8 P$ j- Y) x最近一段时间窗口内的数据。
+ S- v; l& U0 f. ^2 L, B13.4.1　原理
流式计算强调的是数据流的实时性。MapReduce系统主要解决的是对静态数据的
8 m# R2 K1 K7 P1 o$ T: `3 q7 W0 ]批量处理，当MapReduce作业启动时，已经准备好了输入数据，比如保存在分布式文
& {% w  R% F: o8 M( [8 z% @# U件系统上。而流式计算系统在启动时，输入数据一般并没有完全到位，而是经由外
部数据流源源不断地流入。另外，流式计算并不像批处理系统那样，重视数据处理
的总吞吐量，而是更加重视对数据处理的延迟。
MapReduce及其扩展采用的是一种比较静态的模型，如果用它来做数据流的处
  C4 U% U' ~. r理，首先需要将数据流缓存并分块，然后放入集群计算。如果MapReduce每次处理的
: S) y' e' d! @数据量较小，缓存数据流的时间较短，但是，MapReduce框架造成的额外开销将会占
很大比重；如果MapReduce每次处理的数据量较大，缓存数据流的时间会很长，无法
满足实时性的要求。
流式计算系统架构如图13-5所示。
图　13-5　流式计算系统
) J0 ?. }( B8 V  }% {" ~源数据写入到流处理节点，流处理节点内部运行用户自定义的钩子函数对输入
流进行处理，处理完后根据一定的规则转发给下游的流处理节点继续处理。另外，
4 A+ r. U4 m: v% ]9 S系统中往往还有管理节点，用来管理流处理节点的状态以及节点之间的路由规则。
典型钩子函数包括：
+ O+ u0 D( E0 i. W2 Y●聚合函数：计算最近一段时间窗口内数据的聚合值，如max、min、avg、sum、
, c* j. w9 W9 a5 Bcount等。
" f" \0 D$ x7 Y/ v●过滤函数：过滤最近一段时间窗口内满足某些特性的数据，如过滤1秒钟内重
  }7 K5 n- T8 P- @/ |9 R复的点击。
5 p& I; g% d4 ]# y* }2 t如果考虑机器故障，问题变得复杂。上游的处理节点出现故障时，下游有两种
9 Q7 C$ V8 ]# K  b& M& F3 Z1 t选择：第一种选择是等待上游恢复服务，保证数据一致性；第二种选择是继续处
  ~" H2 O% A. T. l6 T3 J理，优先保证可用性，等到上游恢复后再修复错误的计算结果。
  h( O; a& s: }/ u8 ?; u' V5 p流处理节点可以通过主备同步（Master/Slave）的方式容错，即将数据强同步到
0 a  B2 T5 A. M7 G+ ^备机，如果主机出现故障，备机自动切换为主机继续提供服务。然而，这种方式的
代价很高，且流式处理系统往往对错误有一定的容忍度，实际应用时经常选择其他
代价更低的容错方式。
13.4.2　Yahoo S4
4 E. U' m0 ^& c! IYahoo S4最初是Yahoo为了提高搜索广告有效点击率而开发的一个流式处理系
* u  w! U" [' x3 h. s. X统。S4的主要设计目标是提供一种简单的编程接口来处理数据流，使得用户可以定
3 ?  `& @. `/ u( r制流式计算的操作算子。在容错设计上，S4做得比较简单：一旦S4集群中的某个节
点故障，会自动切换到另外一个备用节点，但是原节点的内存状态将丢失。这种方
7 N; o; `$ h  q1 d- u式虽然可能丢失一部分数据，但是成本较低。考虑到服务器故障的概率很低，能够
4 _7 a& j4 A4 y/ [很好地满足流式计算业务需求。
) v( d2 h" ~0 I6 m( g  RS4中每个流处理节点称为一个处理节点（Processing Node,PN），其主要工作是
监听事件，当事件到达时调用合适的处理元（Processing Elements,PE）处理事件。如
果PE有输出，则还需调用通信层接口进行事件的分发和输出，如图13-6所示。
/ s5 d! G8 ?4 l& I0 |- z图　13-6　S4处理节点内部模块
事件监听器（Event Listener）负责监听事件并转交给PE容器（Processing Element
7 R7 b: r) B  xContainer,PEC），由PEC交给合适的PE处理业务逻辑。配置文件中会配置PE原型
, L8 o* {8 L- g  c（PE prototype），包括其功能、处理的事件类型（event type）、关心的key以及关心
' [6 ~- t% K1 g* d) H的key值。每个PE只负责处理自己所关心的事件，也就是说，只有当事件类型、key
! `+ y( t4 s6 R; ?  M类型和key值都匹配时，才会交由该PE进行计算处理。PE处理完逻辑后根据其定义的
- X! Z$ i1 k  \  U* b输出方法可以输出事件，事件交由分发器（Dispatcher）与通信层（Communication
, w) E) ?$ L1 L4 _Layer）进行交互并由输出器（Emitter）输出至下一个逻辑节点。输出器通过对事件
的类型、key类型、key值计算哈希值，以路由到配置文件中指定的PN。
通信层提供集群路由（Routing）、负载均衡（Load Balancing）、故障恢复管理
) D; p" X( T& I" m3 p! E4 i（Failover Management）、逻辑节点到物理节点的映射（存放在Zookeeper上）。当检
测到节点故障时，会切换到备用节点，并自动更新映射关系。通信层隐藏的映射使
- [; y3 i% E* Z) a4 s/ M得PN发送消息时只需要关心逻辑节点而不用关心物理节点。
13.4.3　Twitter Storm
% f  G6 l' b! `: K0 [0 XTwitter Storm是目前广泛使用的流式计算系统，它创造性地引入了一种记录级容
: U: [% h8 w! N/ k! Q错的方法。如图13-7所示，Storm系统中包含如下几种角色：
图　13-7　Storm集群的基本组件
2 @- t! g6 |% _8 ]) w4 l●Nimbus：负责资源分配、任务调度、监控状态。Nimbus和supervisor之间的所
有协调工作都是通过一个Zookeeper集群来完成。
●Supervisor：负责接受nimbus分配的任务，启动和停止属于自己管理的Worker进
程。
+ j7 s- \4 l$ y+ o' {9 L3 e- l●Worker：运行spout/bolt组件的进程。
●Spout：产生源数据流的组件。通常情况下spout会从外部数据源中读取数据，
4 N4 E  O# ]+ g/ l4 V6 l1 v! m然后转换为内部的数据格式。Spout是一个主动的角色，其接口中有个nextTuple()函
$ |7 z5 u. r7 O% c# z数，Storm框架会不停地调用此函数，用户只要在其中生成源数据即可。
●Bolt：接受数据然后执行处理的组件。Bolt可以执行过滤、函数操作、合并、
& U; F/ X5 V+ \0 S写数据库等任何操作。Bolt是一个被动的角色，其接口中有个execute（Tuple input）
函数，在接受到消息后会调用此函数，用户可以在其中执行自己想要的操作。
每个worker上运行着spolt或者bolt组件，数据从spolt组件流入，经过一系列bolt
组件的处理直到生成用户想要的结果。
Storm中的一个记录称为tuple，用户在spout中生成一个新的源tuple时可以为其指
定一个消息编号（message id），多个源tuple可以共用一个消息编号，表示这多个源
tuple对用户来说是同一个消息单元。Storm的记录级容错机制会告知用户由Spolt发出
的每个消息单元是否在指定时间内被完全处理了，从而允许Splot重新发送出错的消
7 o) H, H, W- f息。如图13-8，message1绑定的源tuple1和tuple2经过了bolt1和bolt2的处理后生成两个
新的tuple（tuple3和tuple4），并最终都流向bolt3。当这个过程全部完成时，message1
  X+ O+ {. Q- e$ Y8 j被完全处理了。Storm中有一个系统级组件，叫做acker。这个acker的任务就是追踪从
spout中流出来的每一个message绑定的若干tuple的处理路径。Bolt1、bolt2、bolt3每次
( x& B- x/ D/ j8 h/ x: R处理完成一个tuple都会通知acker,acker会判断message1是否被完全处理了，等到完全
' b2 C+ E* ~' ~/ r处理时通知生成message1的spolt。这里存在两个问题：
3 F3 Q" V8 l, Q. N1 Z+ n# ^0 t图　13-8　Storm数据流示例
1）如何判断message1是否被完全处理了？
2 q* J# k+ L: v% F! A6 `- RAcker中保存了message1对应的校验值（64位整数），初始为0。每次发送或者接
) L2 P5 R) v5 q$ w( b收一个message1绑定的tuple时都会将tuple的编号与校验值进行异或（XOR）运行，如
果每个发送出去的tuple都被接受了，那么，message1对应校验值一定是0，从而认为
( O, Z1 I8 _6 {' t% D9 w% C, c/ `message1被完全处理了。当然，这种方式有一定的误判率，然而考虑到每个tuple的编
号为64位整数，这种概率很低。
2）系统中有很多acker实例，如何选择将message1发给哪个实例？
Storm中采用一致性哈希算法来计算message1对应的acker实例。如果acker出现性
2 B/ |: q/ I% m( n; G8 ?7 v3 [能瓶颈，只需要往系统中加入新的acker实例即可。
2 V& ]$ Y/ p4 z! U
8 v! P* |4 C: l* D! G
6 a- @7 ~7 l: S# w. f