为什么说流处理即未来？

本文整理自Stephan Ewen在Flink Forward China 2018 上的演讲《Stream Processing takes on Everything》。

大家好，今天我的演讲主题看似比较激进：流处理解决所有问题。从某种程度上来说，你可以认为这个演讲是之前晓伟对Flink未来（详情参见上一篇文章：Apache Flink®- 重新定义计算）描述的一个继续。很多人可能对Flink还是停留在最初的认知，觉得Flink是一个流处理引擎，实际上Flink可以做很多其他的工作，比如批处理，比如应用程序。接下来我会简单的说明我对Flink功能的观点，然后我会深入介绍一个特别领域的应用和事件处理场景。这个场景乍看起来不是一个流处理的使用场景，但是在我看来，实际上它就是一个很有趣的流处理使用场景。

上图对为什么流处理可以处理一切作出诠释，将数据看做流是一个自然而又十分强大的想法。大部分数据的产生过程都是随时间生成的流，比如一个Petabyte的数据不会凭空产生。这些数据通常都是一些事件的积累，比如支付、将商品放入购物车，网页浏览，传感器采样输出。

基于数据是流的想法，我们对数据处理可以有相应的理解。比如将过去的历史数据看做是一个截止到某一时刻的有限的流，或是将一个实时处理应用看成是从某一个时刻开始处理未来到达的数据。可能在未来某个时刻它会停止，那么它就变成了处理从开始时刻到停止时刻的有限数据的批处理。当然，它也有可能一直运行下去，不断处理新到达的数据。这个对数据的重要理解方式非常强大，基于这一理解，Flink可以支持整个数据处理范畴内的所有场景。

最广为人知的Flink使用场景是流分析、连续处理（或者说渐进式处理），这些场景中Flink实时或者近实时的处理数据，或者采集之前提到的历史数据并且连续的对这些事件进行计算。晓伟在之前的演讲中提到一个非常好的例子来说明怎么样通过对Flink进行一些优化，进而可以针对有限数据集做一些特别的处理，这使得Flink能够很好的支持批处理的场景，从性能上来说能够与最先进的批处理引擎相媲美。而在这根轴的另一头，是我今天的演讲将要说明的场景 – 事件驱动的应用。这类应用普遍存在于任何服务或者微服务的架构中。这类应用接收各类事件（可能是RPC调用、HTTP请求），并且对这些事件作出一些响应，比如把商品放进购物车，或者加入社交网络中的某个群组。

在我进一步展开今天的演讲之前，我想先对社区在Flink的传统领域（实时分析、连续处理）近期所做的工作做一个介绍。Flink 1.7在2018年11月30日已经发布。在Flink 1.7中为典型的流处理场景加入了一些非常有趣的功能。比如我个人非常感兴趣的在流式SQL中带时间版本的Join。一个基本想法是有两个不同的流，其中一个流被定义为随时间变化的参照表，另一个是与参照表进行Join的事件流。比如事件流是一个订单流，参照表是不断被更新的汇率，而每个订单需要使用最新的汇率来进行换算，并将换算的结果输出到结果表。这个例子在标准的SQL当中实际上并不容易表达，但在我们对Streaming SQL做了一点小的扩展以后，这个逻辑表达变得非常简单，我们发现这样的表达有非常多的应用场景。

另一个在流处理领域十分强大的新功能是将复杂事件处理（CEP）和SQL相结合。CEP应用观察事件模式。比如某个CEP应用观察股市，当有两个上涨后紧跟一个下跌时，这个应用可能做些交易。再比如一个观察温度计的应用，当它发现有温度计在两个超过90摄氏度的读数之后的两分钟里没有任何操作，可能会进行一些操作。与SQL的结合使这类逻辑的表达也变得非常简单。

第三个Flink 1.7中做了很多工作的功能是Schema升级。这个功能和基于流的应用紧密相关。就像你可以对数据库进行数据Schema升级一样，你可以修改Flink表中列的类型或者重新写一个列。

另外我想简单介绍的是流处理技术不仅仅是简单对数据进行计算，这还包括了很多与外部系统进行事务交互。流处理引擎需要在采用不同协议的系统之间以事务的方式移动数据，并保证计算过程和数据的一致性。这一部分功能也是在Flink 1.7中得到了增强。

以上我对Flink 1.7的新功能向大家做了简单总结。下面让我们来看看今天我演讲的主要部分，也就是利用Flink来搭建应用和服务。我将说明为什么流处理是一个搭建应用和服务或者微服务的有趣技术。

我将从左边这个高度简化的图说起，我们一会儿将聊一些其中的细节。首先我们来看一个理解应用简单的视角。如左图所示，一个应用可以是一个Container，一个Spring应用，或者Java应用、Ruby应用，等等。这个应用从诸如RPC，HTTP等渠道接收请求，然后依据请求进行数据库变更。这个应用也可能调用另一个微服务并进行下一步的处理。我们可以非常自然的想到进入到应用的这些请求可以看做是个事件组成的序列，所以我们可以把它们看做是事件流。可能这些事件被缓存在消息队列中，而应用会从消息队列中消费这些事件进行处理，当应用需要响应一个请求时，它将结果输出到另一个消息队列，而请求发送方可以从这个消息队列中消费得到所发送请求的响应。在这张图中我们已经可以看到一些有趣的不同。

第一个不同是在这张图中应用和数据库不再是分开的两个实体，而是被一个有状态的流处理应用所代替。所以在流处理应用的架构中，不再有应用和数据库的连接了，它们被放到了一起。这个做法有利有弊，但其中有些好处是非常重要的。首先是性能上的好处是明显的，因为应用不再需要和数据库进行交互，处理可以基于内存中的变量进行。其次这种做法有很好并且很简单的一致性。

这张图被简化了很多，实际上我们通常会有很多个应用，而不是一个被隔离的应用，很多情况下你的应用会更符合这张图。系统中有个接收请求的接口，然后请求被发送到第一个应用，可能会再被发到另一个应用，然后得到相应。在图中有些应用会消费中间结果的流。这张图已经展示了为什么流处理是更适合比较复杂的微服务场景的技术。因为很多时候系统中不会有一个直接接收用户请求并直接响应的服务，通常来说一个微服务需要跟其他微服务通信。这正如在流处理的架构中不同应用在创建输出流，同时基于衍生出的流再创建并输出新的流。

到目前为止，我们看到的内容多少还比较直观。而对基于流处理技术的微服务架构而言，人们最常问的一个问题是如何保证事务性？如果系统中使用的是数据库，通常来说都会有非常成熟复杂的数据校验和事务模型。这也是数据库在过去许多年中十分成功的原因。开始一个事务，对数据做一些操作，提交或者撤销一个事务。这个机制使得数据完整性得到了保证（一致性，持久性等等）。

那么在流处理中我们怎么做到同样的事情呢？作为一个优秀的流处理引擎，Flink支持了恰好一次语义，保证了每个事件只会被处理一遍。但是这依然对某些操作有限制，这也成为了使用流处理应用的一个障碍。我们通过一个非常简单流处理应用例子来看我们可以做一些什么扩展来解决这个问题。我们会看到，解决办法其实出奇的简单。

让我们以这个教科书式的事务为例子来看一下事务性应用的过程。这个系统维护了账户和其中存款余额的信息。这样的信息可能是银行或者在线支付系统的场景中用到的。假设我们想要处理类似下面的事务：

如果账户A中的余额大于100，那么从账户A中转账50元到账户B。

这是个非常简单的两个账户之间进行转账的例子。

数据库对于这样的事务已经有了一个核心的范式，也就是原子性，一致性，隔离性和持久性（ACID）。这是能够让用户放心使用事务的几个基本保证。有了他们，用户不用担心钱在转账过程中会丢失或者其他问题。让我们用这个例子来放到流处理应用中，来让流处理应用也能提供和数据相同的ACID支持：

原子性要求一个转账要不就完全完成，也就是说转账金额从一个账户减少，并增加到另一个账户，要不就两个账户的余额都没有变化。而不会只有一个账户余额改变。否则的话钱就会凭空减少或者凭空增加。

一致性和隔离性是说如果有很多用户同时想要进行转账，那么这些转账行为之间应该互不干扰，每个转账行为应该被独立的完成，并且完成后每个账户的余额应该是正确的。也就是说如果两个用户同时操作同一个账户，系统不应该出错。

持久性指的是如果一个操作已经完成，那么这个操作的结果会被妥善的保存而不会丢失。

我们假设持久性已经被满足。一个流处理器有状态，这个状态会被checkpoint，所以流处理器的状态是可恢复的。也就是说只要我们完成了一个修改，并且这个修改被checkpoint了，那么这个修改就是持久化的。

让我们来看看另外三个例子。设想一下，如果我们用流处理应用来实现这样一个转账系统会发生什么。我们先把问题简化一些，假设转账不需要有条件，仅仅是将50元从账户A转到账户，也就是说账户A的余额减少50元而账户B的余额增加50元。我们的系统是一个分布式的并行系统，而不是一个单机系统。简单起见我们假设系统中只有两台机器，这两台机器可以是不同的物理机或者是在YARN或者Kubernetes上不同的容器。总之它们是两个不同的流处理器实例，数据分布在这两个流处理器上。我们假设账户A的数据由其中一台机器维护，而账户B的数据有另一台机器维护。

现在我们要做个转账，将50元从账户A转移到账户B，我们把这个请求放进队列中，然后这个转账请求被分解为对账户A和B分别进行操作，并且根据键将这两个操作路由到维护账户A和维护账户B的这两台机器上，这两台机器分别根据要求对账户A和账户B的余额进行改动。这并不是事务操作，而只是两个独立无意义的改动。一旦我们将转账的请求改的稍微复杂一些就会发现问题。

下面我们假设转账是有条件的，我们只想在账户A的余额足够的情况下才进行转账，这样就已经有些不太对了。如果我们还是像之前那样操作，将这个转账请求分别发送给维护账户A和B的两台机器，如果A没有足够的余额，那么A的余额不会发生变化，而B的余额可能已经被改动了。我们就违反了一致性的要求。

我们看到我们需要首先以某种方式统一做出是否需要更改余额的决定，如果这个统一的决定中余额需要被修改，我们再进行修改余额的操作。所以我们先给维护A的余额的机器发送一个请求，让它查看A的余额。我们也可以对B做同样的事情，但是这个例子里面我们不关心B的余额。然后我们把所有这样的条件检查的请求汇总起来去检验条件是否满足。因为Flink这样的流处理器支持迭代，如果满足转账条件，我们可以把这个余额改动的操作放进迭代的反馈流当中来告诉对应的节点来进行余额修改。反之如果条件不满足，那么余额改动的操作将不会被放进反馈流。这个例子里面，通过这种方式我们可以正确的进行转账操作。从某种角度上来说我们实现了原子性，基于一个条件我们可以进行全部的余额修改，或者不进行任何余额修改。这部分依然还是比较直观的，更大的困难是在于如何做到并发请求的隔离性。

假设我们的系统没有变，但是系统中有多个并发的请求。我们在之前的演讲中已经知道，这样的并发可能达到每秒钟几十亿条。如图，我们的系统可能从两个流中同时接受请求。如果这两个请求同时到达，我们像之前那样将每个请求拆分成多个请求，首先检查余额条件，然后进行余额操作。然而我们发现这会带来问题。管理账户A的机器会首先检查A的余额是否大于50，然后又会检查A的余额是否大于100，因为两个条件都满足，所以两笔转账操作都会进行，但实际上账户A上的余额可能无法同时完成两笔转账，而只能完成50元或者100元的转账中的一笔。这里我们需要进一步思考怎么样来处理并发的请求，我们不能只是简单地并发处理请求，这会违反事务的保证。从某种角度来说，这是整个数据库事务的核心。数据库的专家们花了一些时间提供了不同解决方案，有的方案比较简单，有的则很复杂。但所有的方案都不是那么容易，尤其是在分布式系统当中。

在流处理中怎么解决这个问题呢？直觉上讲，如果我们能够让所有的事务都按照顺序依次发生，那么问题就解决了，这也被成为可序列化的特性。但是我们当然不希望所有的请求都被依次顺序处理，这与我们使用分布式系统的初衷相违背。所以我们需要保证这些请求最后的产生的影响看起来是按照顺序发生的，也就是一个请求产生的影响是基于前一个请求产生影响的基础之上的。换句话说也就是一个事务的修改需要在前一个事务的所有修改都完成后才能进行。这种希望一件事在另一件事之后发生的要求看起来很熟悉，这似乎是我们以前在流处理中曾经遇到过的问题。是的，这听上去像是事件时间。用高度简化的方式来解释，如果所有的请求都在不同的事件时间产生，即使由于种种原因他们到达处理器的时间是乱序的，流处理器依然会根据他们的事件时间来对他们进行处理。流处理器会使得所有的事件的影响看上去都是按顺序发生的。按事件时间处理是Flink已经支持的功能。

那么详细说来，我们到底怎么解决这个一致性问题呢？假设我们有并行的请求输入并行的事务请求，这些请求读取某些表中的记录，然后修改某些表中的记录。我们首先需要做的是把这些事务请求根据事件时间顺序摆放。这些请求的事务时间不能够相同，但是他们之间的时间也需要足够接近，这是因为在事件时间的处理过程中会引入一定的延迟，我们需要保证所处理的事件时间在向前推进。因此第一步是定义事务执行的顺序，也就是说需要有一个聪明的算法来为每个事务制定事件时间。在图上，假设这三个事务的事件时间分别是T+2, T和T+1。那么第二个事务的影响需要在第一和第三个事务之前。不同的事务所做的修改是不同的，每个事务都会产生不同的操作请求来修改状态。我们现在需要将对访问每个行和状态的事件进行排序，保证他们的访问是符合事件时间顺序的。这也意味着那些相互之间没有关系的事务之间自然也没有了任何影响。比如这里的第三个事务请求，它与前两个事务之间没有访问共同的状态，所以它的事件时间排序与前两个事务也相互独立。而当前两个事务之间的操作的到达顺序与事件时间不符时，Flink则会依据它们的事件时间进行排序后再处理。

必须承认，这样说还是进行了一些简化，我们还需要做一些事情来保证高效执行，但是总体原则上来说，这就是全部的设计。除此之外我们并不需要更多其他东西。

为了实现这个设计，我们引入了一种聪明的分布式事件时间分配机制。这里的事件时间是逻辑时间，它并不需要有什么现实意义，比如它不需要是真实的时钟。使用Flink的乱序处理能力，并且使用Flink迭代计算的功能来进行某些前提条件的检查。这些就是我们构建一个支持事务的流处理器的要素。

我们实际上已经完成了这个工作，称之为流式账簿（Streaming Ledger），这是个在Apache Flink上很小的库。它基于流处理器做到了满足ACID的多键事务性操作。我相信这是个非常有趣的进化。流处理器一开始基本上没有任何保障，然后类似Storm的系统增加了至少一次的保证。但显然至少一次依然不够好。然后我们看到了恰好一次的语义，这是一个大的进步，但这只是对于单行操作的恰好一次语义，这与键值库很类似。而支持多行恰好一次或者多行事务操作将流处理器提升到了一个可以解决传统意义上关系型数据库所应用场景的阶段。

Streaming Ledger的实现方式是允许用户定义一些表和对这些表进行修改的函数。Streaming Ledger会运行这些函数和表，所有的这些一起编译成一个Apache Flink的有向无环图（DAG）。Streaming Ledger会注入所有事务时间分配的逻辑，以此来保证所有事务的一致性。

搭建这样一个库并不难，难的是让它高性能的运行。让我们来看看它的性能。这些性能测试是几个月之前的，我们并没有做什么特别的优化，我们只是想看看一些最简单的方法能够有什么样的性能表现。而实际性能表现看起来相当不错。如果你看这些性能条形成的阶梯跨度，随着流处理器数量的增长，性能的增长相当线性。在事务设计中，没有任何协同或者锁参与其中。这只是流处理，将事件流推入系统，缓存一小段时间来做一些乱序处理，然后做一些本地状态更新。在这个方案中，没有什么特别代价高昂的操作。在图中性能增长似乎超过了线性，我想这主要是因为JAVA的JVM当中GC的工作原因导致的。在32个节点的情况下我们每秒可以处理大约两百万个事务。为了与数据库性能测试进行对比，通常当你看数据库的性能测试时，你会看到类似读写操作比的说明，比如10%的更新操作。而我们的测试使用的是100%的更新操作，而每个写操作至少更新在不同分区上的4行数据，我们的表的大小大约是两亿行。即便没有任何优化，这个方案的性能也非常不错。

另一个在事务性能中有趣的问题是当更新的操作对象是一个比较小的集合时的性能。如果事务之间没有冲突，并发的事务处理是一个容易的事情。如果所有的事务都独立进行而互不干扰，那这个不是什么难题，任何系统应该都能很好的解决这样的问题。当所有的事务都开始操作同一些行时，事情开始变得更有趣了，你需要隔离不同的修改来保证一致性。所以我们开始比较一个只读的程序、一个又读又写但是没有写冲突的程序和一个又读又写并有中等程度写冲突的程序这三者之间的性能。你可以看到性能表现相当稳定。这就像是一个乐观的并发冲突控制，表现很不错。那如果我们真的想要针对这类系统的阿喀琉斯之踵进行考验，也就是反复的更新同一个小集合中的键。在传统数据库中，这种情况下可能会出现反复重试，反复失败再重试，这是一种我们总想避免的糟糕情况。是的，我们的确需要付出性能代价，这很自然，因为如果你的表中有几行数据每个人都想更新，那么你的系统就失去了并发性，这本身就是个问题。但是这种情况下，系统并没崩溃，它仍然在稳定的处理请求，虽然失去了一些并发性，但是请求依然能够被处理。这是因为我们没有冲突重试的机制，你可以认为我们有一个基于乱序处理天然的冲突避免的机制，这是一种非常稳定和强大的技术。

我们还尝试了在跨地域分布的情况下的性能表现。比如我们在美国、巴西，欧洲，日本和澳大利亚各设置了一个Flink集群。也就是说我们有个全球分布的系统。如果你在使用一个关系型数据库，那么你会付出相当高昂的性能代价，因为通信的延迟变得相当高。跨大洲的信息交互比在同一个数据中心甚至同一个机架上的信息交互要产生大得多的延迟。但是有趣的是，流处理的方式对延迟并不是十分敏感，延迟对性能有所影响，但是相比其它很多方案，延迟对流处理的影响要小得多。所以，在这样的全球分布式环境中执行分布式程序，的确会有更差的性能，部分原因也是因为跨大洲的通信带宽不如统一数据中心里的带宽，但是性能表现依然不差。实际上，你可以拿它当做一个跨地域的数据库，同时仍然能够在一个大概10个节点的集群上获得每秒几十万条事务的处理能力。在这个测试中我们只用了10个节点，每个大洲两个节点。所以10个节点可以带来全球分布的每秒20万事务的处理能力。我认为这是很有趣的结果，这是因为这个方案对延迟并不敏感。

我已经说了很多利用流处理来实现事务性的应用。可能听起来这是个很自然的想法，从某种角度上来说的确是这样。但是它的确需要一些很复杂的机制来作为支撑。它需要一个连续处理而非微批处理的能力，需要能够做迭代，需要复杂的基于事件时间处理乱序处理。为了更好地性能，它需要灵活的状态抽象和异步checkpoint机制。这些是真正困难的事情。这些不是由Ledger Streaming库实现的，而是Apache Flink实现的，所以即使对这类事务性的应用而言，Apache Flink也是真正的中流砥柱。

至此，我们可以说流处理不仅仅支持连续处理、流式分析、批处理或者事件驱动的处理，你也可以用它做事务性的处理。当然，前提是你有一个足够强大的流处理引擎。这就是我演讲的全部内容。