扩展为集群

随着业务的增长，数据量越来越大，单一机器逐渐不能满足日益增长的数据量。

与此同时，数据量变大后，程序的性能也开始变得越来越差，以至于最后不可接受。所以，我们必须未雨绸缪，让流计算系统能够伴随业务不断成长，这就要求系统具备水平扩展的能力。

在实时流计算系统中，不管是使用诸如Kafka消息中间件的分区功能，还是依赖于诸如Flink KeyedStream这样的流计算框架本身的分区流能力，最终都能比较轻松、方便地实现计算能力的水平扩展。

但是，对于计算中的状态数据来说，实现计算能力的水平扩展不是一件非常容易的事情。这是因为，状态数据很多时候是需要共享和同步的，如对于分别在两个计算节点上计算的事件，它们可能需要同时访问相同的数据。即使我们先不考虑并发安全的问题，这也意味着相同的数据会被两个不同的节点访问。也就是说，至少有一个节点的跨网络远程访问是不可避免的。

而在前面关于时间维度聚合特征计算和关联图谱特征计算的具体实现中，我们不难发现，它们都是严重依赖于大量状态访问的，甚至有时候一次窗口计数的查询会访问几个甚至几十个子窗口的寄存器。如果不能避免或优化这些访问，那么程序的性能势必会严重受累于跨网络的远程状态访问。所以，我们有必要专门讨论将状态的存储和管理，从单节点扩展为分布集群时的一系列问题。本节将讨论3种不同的状态集群方案，它们分别代表了一种典型的分布式计算架构设计思路，可谓是各有千秋。

基于Redis的状态集群

图5-2展示了使用Redis集群实现状态分布式存储和管理的原理。

当采用Redis集群实现分布式状态存储和管理时，流计算集群和Redis集群节点是分离开的。流计算集群中的每个节点都可以任意访问Redis集群中的任何一个节点。这样的架构有一个非常明显的好处，即计算和数据是分离开的。我们在任何时候，可以任意地新增流计算节点，而不会影响Redis集群。反过来，我们也可以任意地新增Redis节点，而不会影响流计算集群。

这样的体系结构也有缺点。以“过去一天同一用户的总交易金额”这个时间维度聚合特征的计算为例。如果我们采用与上面相同的算法，就需要先将“1天”分成了24个“1小时”的子窗口，这样在查询计算时将有24次的Redis GET操作。假设这24个子窗口的状态数据是分散在6台Redis上的。如果不做任何优化设计，那么这一个特征计算就需要24次I/O操作，而且涉及与6台不同服务器的远程通信，这势必会对性能造成极大的影响。针对以上问题，我们该怎么办呢？我们可以根据“局部性原理”和“批次请求处理”的思想来优化解决方案。

局部性原理

局部性原理（见图5-3）是指计算单元在访问存储单元时，所访问的存储单元应该趋向于聚集在一个局部的连续区域内。利用局部性原理可以更加充分地提高计算资源的使用效率，从而提高程序的性能。

前面讲到在“过去一天同一用户的总交易金额”这个特征计算中，我们可能需要访问6台Redis节点上的数据。这是因为默认情况下，Redis集群将数据按照key做hash计算后分散到各个槽（slot）中，而槽又分布在各个Redis节点上。如果我们能够让“同一用户”的状态数据保存在相同的槽里，就可以让这批数据存在于相同的Redis节点上。

Redis的官方集群方案Redis Cluster提供了贴心的标签（tag）功能，允许只使用key中的部分字段来计算hash值。具体而言，如果hash_tag指定为“{}”，那么当key含有“{}”的时候，就不使用整个key来计算hash值，而只对“{}”包括的部分字段计算hash值。例如，在使用标签功能后，每个小窗口内记录这个窗口交易总金额的key如下所示：

$event_type.{$userid}.$window_unit.$window_index

经过标签化的key，相同用户的状态数据会落在相同的Redis节点。

这样，我们只需要访问一个Redis节点即可。

现在数据位于同一个节点上了，那这有什么好处呢？好处多着呢。

首先，我们可以放心大胆地使用Redis的各种多键指令了，如MGET、MSET、SUNION和SUNIONSTORE等。这些指令在操作过程中可以一次访问多个键，从而提高指令执行效率。而如果这些key不在同一个Redis节点上，则这些指令是不能使用的。其次，我们可以充分发挥Redis的pipeline功能。通过Redis的pipeline功能，可以一次性发送多条指令，这些指令间可以没有任何依赖关系。当执行完后，这些指令的结果一次性返回。通过这种批次传递和执行指令的方式，Redis减少了平均每条指令执行时不必要的网络开销，提升了执行效率。同样地，如果这些数据不在同一个Redis节点上，我们就不能使用pipeline功能。

所以，将相关数据放在相同的节点上，给我们留下更多的优化空间。经过上述的优化设计后，原本需要24次I/O操作的特征计算，最优情况下只需要一次I/O操作。这就是局部性原理的魅力所在！

当然，使用局部性原理也可能出现数据在集群节点上分布不均匀的问题。所以，在选择分区标签时，应该尽量分得更细、更均匀些，这样可以减小数据倾斜的问题。

批次请求处理

批次请求处理是指将多个请求收集起来后，一次性成批处理的过程。批次请求处理可以降低均摊在每条消息处理时非有效用于消息处理的资源和时间。Redis的pipeline功能就是一种批次请求处理的技术，但是我们不能仅限于Redis的Pipeline功能。实际上，任何与I/O相关的操作都可以借鉴这种批次处理的思想，如RPC（远程过程调用）、REST请求、数据库查询等。

在实际开发过程中，对请求做批次化处理本身并不是非常复杂的过程，比较麻烦的是应该怎样将分布在程序各个地方的请求收集起来。针对这个问题，我们可以使用队列和CompletableFuture的异步方案，图5-4描述了这个方案的具体实现方法。

当请求发起时，将请求提交给队列后获取一个CompletableFuture对象。而另外一个线程等着从这个队列中取出请求。当该线程取出的请求达到一定数量或者等待超过一定时间时，将取出的这批请求封装成批次请求，发送给请求处理服务器。当批次请求返回后，将批次结果拆解开，再依次使用CompletableFuture的complete函数将结果交给各个请求发起者。这样就实现了请求的批次化处理。

批次化处理的好处在于提高了请求处理的吞吐量，降低了每条请求平均响应时延，但是因为使用了队列和异步的方案，也有可能会提高特定某条请求的响应时延。因此，在实际开发中，读者需要综合考虑自己的场景选择最合适的方案。

基于Apache Ignite的状态集群

图5-5描述了Apache Ignite集群用于状态存储和管理的架构。从该架构图可以看出，当采用Apache Ignite来实现状态管理时，计算节点和数据节点是耦合在一起的，它们在相同的JVM内运行。每个ApacheIgnite节点会保存全部集群数据中的一部分，流计算节点通过其嵌入的Apache Ignite节点来访问状态数据。而Apache Ignite数据格点自身的设计和实现机制，允许计算尽量只需要访问本地节点上的数据以完成计算任务，减少数据在网络间的流动。这种设计方案充分利用了ApacheIgnite提供的数据格点能力，是一种典型的网格计算架构。

采用Apache Ignite数据格点的方案，可以让我们不必过多考虑数据分区问题。Apache Ignite会自行处理数据局部性及计算和数据亲和性的问题。Apach.Ignite提供的各种计算和查询接口屏蔽了分布式数据和分布式计算的复杂性，为我们开发分布式系统带来极大的便利性。网格计算中的所有节点都是平等的，当需要水平扩展集群时，只需要将新的节点添加到网格中即可。

不过将计算节点和数据节点耦合在同一个JVM后，增加了单一节点的复杂性，同时使计算资源的分配、管理和监控等变得更加复杂。这点需要读者在做方案选型时根据具体场景和需求自行定夺。

基于分布式文件系统的状态管理集群

除了上面两种状态管理的集群外，还有一种基于分布式文件系统的状态管理集群，这是一种非常典型的分布式状态管理方案。Flink的状态管理采用的就是这种方案。

图5-6描述了基于分布式文件系统的状态管理集群。在这种分布式状态管理方案中，流计算节点针对状态的操作完全在本地完成，不涉及任何远程操作。但如果只是这样，那当需要扩展或收缩集群的节点数时，怎么保证能够读取到原来的状态信息呢？因此在每个节点上，有专门的线程定期或在必要的时候（如任务关闭前），对状态进行checkpoint。所谓checkpoint，是指将本地状态后端的数据做快照（snapshot）之后，保存到分布式文件系统的过程。当集群在节点数变化后再重启时，各个节点首先从分布式文件系统中读取其所负责数据分片所在的快照，再将快照恢复到状态后端，这样各个节点就获得重启前的状态数据了，之后的计算又可以完全在本地完成。

这种方案的优势在于，流计算节点对状态的操作在本地完成，不需要任何远程操作。这样本地状态后端的选择可以非常丰富，给性能优化留下极大空间。例如，Flink目前已经支持内存、文件系统和RockDB 3种状态后端。不过这种方案也有一个缺点，即不能在运行时动态扩展或缩小集群。当集群节点数变更时，需要重启集群。对应在Flink中，当需要改变算子的并行度（operator parallelism）时，必须重启作业。

本篇文章给大家讲解的内容是实时流计算的状态管理：扩展为集群

下篇文章给大家讲解的内容是开源流计算框架：Apache Storm