|
|
13.4 流式计算
1 \) y0 S9 R0 L) }6 D, D! d4 d- P& }MapReduce及其扩展解决了离线批处理问题，但是无法保证实时性。对于实时性
# N& |6 ]+ d% d) Q. C# X7 n s要求高的场景，可以采用流式计算或者实时分析系统进行处理。
- a+ A: z/ h* {: W0 H流式计算（Stream Processing）解决在线聚合（Online Aggregation）、在线过滤
, w J/ G' k% q4 v' a9 S% C8 H1 { X' W（Online Filter）等问题，流式计算同时具有存储系统和计算系统的特点，经常应用
+ x/ I1 w2 x( p/ i: A: m* d7 U4 l [在一些类似反作弊、交易异常监控等场景。流式计算的操作算子和时间相关，处理6 z2 @& ]3 o5 t( W7 M1 [
最近一段时间窗口内的数据。
# Y8 {( v3 z$ B! z4 }13.4.1 原理; L) D& M; e- z8 `+ E: y
流式计算强调的是数据流的实时性。MapReduce系统主要解决的是对静态数据的' n$ o6 d' y+ _/ P" c6 J e
批量处理，当MapReduce作业启动时，已经准备好了输入数据，比如保存在分布式文
) G- p. V+ q, i! T件系统上。而流式计算系统在启动时，输入数据一般并没有完全到位，而是经由外- l9 S4 A* A n8 M% [
部数据流源源不断地流入。另外，流式计算并不像批处理系统那样，重视数据处理
+ W- r! E: z* K! R7 p的总吞吐量，而是更加重视对数据处理的延迟。
& F5 v* s# {# s6 SMapReduce及其扩展采用的是一种比较静态的模型，如果用它来做数据流的处
) t! e/ I7 ~0 t( v5 }) D. A: b理，首先需要将数据流缓存并分块，然后放入集群计算。如果MapReduce每次处理的
/ @+ \; T% u& l, b3 ?. j数据量较小，缓存数据流的时间较短，但是，MapReduce框架造成的额外开销将会占. e8 G; w6 h* k+ Z+ l6 u
很大比重；如果MapReduce每次处理的数据量较大，缓存数据流的时间会很长，无法. _8 e( ]- \/ d" A4 j
满足实时性的要求。
& c% T# C; h* C9 g+ C流式计算系统架构如图13-5所示。* G" _+ V: ~0 P6 p& {
图 13-5 流式计算系统% W0 n% l# p4 W; ~' K
源数据写入到流处理节点，流处理节点内部运行用户自定义的钩子函数对输入9 k' X/ ^4 ?0 b1 g! n/ _% J8 L' {* J
流进行处理，处理完后根据一定的规则转发给下游的流处理节点继续处理。另外，! C* R/ d* V" L7 P) Z2 Y4 y( ?
系统中往往还有管理节点，用来管理流处理节点的状态以及节点之间的路由规则。
. T5 d0 q' @( T# P典型钩子函数包括：* {2 u1 ` [9 E
●聚合函数：计算最近一段时间窗口内数据的聚合值，如max、min、avg、sum、) F( q2 \4 y4 Y
count等。( a- D. z4 `) O6 P( @
●过滤函数：过滤最近一段时间窗口内满足某些特性的数据，如过滤1秒钟内重( K' Z$ [: K# i( |- A ?
复的点击。) m3 {) _% i" X4 y- k5 _
如果考虑机器故障，问题变得复杂。上游的处理节点出现故障时，下游有两种; \2 y" T7 X+ `* N% o% q
选择：第一种选择是等待上游恢复服务，保证数据一致性；第二种选择是继续处
, J/ W2 j$ {4 X [3 I理，优先保证可用性，等到上游恢复后再修复错误的计算结果。
$ Q4 r9 P/ A. ?* J* Q流处理节点可以通过主备同步（Master/Slave）的方式容错，即将数据强同步到9 W( R- ] y: F' E
备机，如果主机出现故障，备机自动切换为主机继续提供服务。然而，这种方式的
5 L8 z5 C) D+ C7 k1 o代价很高，且流式处理系统往往对错误有一定的容忍度，实际应用时经常选择其他) O4 A- u! ?; ^
代价更低的容错方式。1 |& P+ d/ v9 |" x- s
13.4.2 Yahoo S4; t: @. q8 z- |4 J" X2 f
Yahoo S4最初是Yahoo为了提高搜索广告有效点击率而开发的一个流式处理系
/ W* I- g6 r! r: ~+ P统。S4的主要设计目标是提供一种简单的编程接口来处理数据流，使得用户可以定
0 r* o* S) l& J- J$ o$ M2 K! A+ a制流式计算的操作算子。在容错设计上，S4做得比较简单：一旦S4集群中的某个节
, I- i5 I* ?+ f! q3 ^! ?7 j点故障，会自动切换到另外一个备用节点，但是原节点的内存状态将丢失。这种方
' Y. V' G& p, m3 W- D! x式虽然可能丢失一部分数据，但是成本较低。考虑到服务器故障的概率很低，能够* z" z5 @0 u" {
很好地满足流式计算业务需求。1 I& S4 H' x. Y( X8 y
S4中每个流处理节点称为一个处理节点（Processing Node,PN），其主要工作是7 Z c! M( ]! B
监听事件，当事件到达时调用合适的处理元（Processing Elements,PE）处理事件。如( p) S. O; d- I* [9 r) \. k1 }
果PE有输出，则还需调用通信层接口进行事件的分发和输出，如图13-6所示。
0 i" e) l9 W+ z Q图 13-6 S4处理节点内部模块
& E" G+ A( v0 U! p事件监听器（Event Listener）负责监听事件并转交给PE容器（Processing Element, i1 S9 |- Z* }9 j# ^- L0 \
Container,PEC），由PEC交给合适的PE处理业务逻辑。配置文件中会配置PE原型
* y$ Z) ?! \, T2 a9 d- s( M（PE prototype），包括其功能、处理的事件类型（event type）、关心的key以及关心
& p/ h; _0 {0 a- m, K的key值。每个PE只负责处理自己所关心的事件，也就是说，只有当事件类型、key$ C, T' @/ I3 L& G9 w
类型和key值都匹配时，才会交由该PE进行计算处理。PE处理完逻辑后根据其定义的# Z1 h. s( V; C2 @, i
输出方法可以输出事件，事件交由分发器（Dispatcher）与通信层（Communication" N) ?- {. V; e! D
Layer）进行交互并由输出器（Emitter）输出至下一个逻辑节点。输出器通过对事件5 ^: T! I3 W5 o. T- ~2 G1 N, E; d( N
的类型、key类型、key值计算哈希值，以路由到配置文件中指定的PN。
8 p, g+ ?- {' V P) A9 y0 K通信层提供集群路由（Routing）、负载均衡（Load Balancing）、故障恢复管理& v9 k& h+ o1 U# j+ T4 ]+ i
（Failover Management）、逻辑节点到物理节点的映射（存放在Zookeeper上）。当检
$ ~0 B: B! D9 C测到节点故障时，会切换到备用节点，并自动更新映射关系。通信层隐藏的映射使) l6 i" B9 c" w( O$ G1 s# M# X
得PN发送消息时只需要关心逻辑节点而不用关心物理节点。0 N- Z# A* k1 i4 a* N! z8 ]+ _* R7 F
13.4.3 Twitter Storm+ s+ j) n% Z! j# a
Twitter Storm是目前广泛使用的流式计算系统，它创造性地引入了一种记录级容0 r( F5 P% G! B
错的方法。如图13-7所示，Storm系统中包含如下几种角色：, K* Q0 A1 [2 B
图 13-7 Storm集群的基本组件
4 x7 Z& U1 j, V& m; g' M●Nimbus：负责资源分配、任务调度、监控状态。Nimbus和supervisor之间的所
3 _ K2 o' ^ h6 m9 c- P有协调工作都是通过一个Zookeeper集群来完成。; ]' @# y5 P7 N+ K3 s
●Supervisor：负责接受nimbus分配的任务，启动和停止属于自己管理的Worker进" T: p; y4 z- J, W5 o
程。8 |& k, X5 x+ A) Q, C# H
●Worker：运行spout/bolt组件的进程。8 R$ ?* ]6 ]3 r
●Spout：产生源数据流的组件。通常情况下spout会从外部数据源中读取数据，, T& h+ k, ]& o0 x
然后转换为内部的数据格式。Spout是一个主动的角色，其接口中有个nextTuple()函; C0 ?$ s2 f J* l! H0 y
数，Storm框架会不停地调用此函数，用户只要在其中生成源数据即可。, P' {# }8 U3 j0 ?9 p
●Bolt：接受数据然后执行处理的组件。Bolt可以执行过滤、函数操作、合并、% ?: ^. Z8 a5 V- R
写数据库等任何操作。Bolt是一个被动的角色，其接口中有个execute（Tuple input）
$ V% @) A( K5 Q0 s/ Y函数，在接受到消息后会调用此函数，用户可以在其中执行自己想要的操作。
- E5 B/ }5 J1 J y: M Q+ k7 J每个worker上运行着spolt或者bolt组件，数据从spolt组件流入，经过一系列bolt+ }$ S5 o/ q& L9 x! e$ M. b, }
组件的处理直到生成用户想要的结果。
) G3 W' a8 ^. j2 h/ d; RStorm中的一个记录称为tuple，用户在spout中生成一个新的源tuple时可以为其指
# E9 i' @! w+ o; f定一个消息编号（message id），多个源tuple可以共用一个消息编号，表示这多个源9 |" l& @, m$ x$ f
tuple对用户来说是同一个消息单元。Storm的记录级容错机制会告知用户由Spolt发出* r7 Q+ g/ N2 `2 T. `; u
的每个消息单元是否在指定时间内被完全处理了，从而允许Splot重新发送出错的消, E) [0 E' C! y. \6 u
息。如图13-8，message1绑定的源tuple1和tuple2经过了bolt1和bolt2的处理后生成两个 q4 Z. Q& {& ?/ S! ?7 i. B
新的tuple（tuple3和tuple4），并最终都流向bolt3。当这个过程全部完成时，message1
, F! e! L2 `# J2 ] A- G, P) x8 F被完全处理了。Storm中有一个系统级组件，叫做acker。这个acker的任务就是追踪从
" F8 U! f# v y' E8 Qspout中流出来的每一个message绑定的若干tuple的处理路径。Bolt1、bolt2、bolt3每次
6 ~4 f4 I( `2 `" M* V$ H8 a处理完成一个tuple都会通知acker,acker会判断message1是否被完全处理了，等到完全
& t; z. b9 K% A3 C/ F处理时通知生成message1的spolt。这里存在两个问题：
% p6 @# x% `/ U; N2 y. v9 b图 13-8 Storm数据流示例
6 V* Q' V) y/ C, F" }$ i3 h1）如何判断message1是否被完全处理了？/ Y) H2 a. r: C
Acker中保存了message1对应的校验值（64位整数），初始为0。每次发送或者接5 d/ }( K$ Z* R( e. a! D
收一个message1绑定的tuple时都会将tuple的编号与校验值进行异或（XOR）运行，如
6 ~) F7 g0 O8 [) \& S果每个发送出去的tuple都被接受了，那么，message1对应校验值一定是0，从而认为
?$ ^2 [6 l2 Xmessage1被完全处理了。当然，这种方式有一定的误判率，然而考虑到每个tuple的编
9 n9 s1 G7 y) V0 Q$ e号为64位整数，这种概率很低。
; [. p' O# ]$ v/ [0 p2）系统中有很多acker实例，如何选择将message1发给哪个实例？
7 S( d% Q+ R' A3 L VStorm中采用一致性哈希算法来计算message1对应的acker实例。如果acker出现性4 A9 m: Z0 p5 n: X8 W
能瓶颈，只需要往系统中加入新的acker实例即可。
3 |% m% K8 r3 z3 s; P% h/ K% |' ^- B
; P' [0 v6 e9 n |
|
发表于 2017-3-20 19:43:57
