阿里实时计算Blink核心技术

实时计算in阿里巴巴

实时计算在阿里巴巴内部应用广泛。随着新经济体的出现与发展，技术的革新和用户需求的提升，人们越来越需要实时计算的能力，它的最大好处就是能够基于实时变化数据更新大数据处理的状态和结果。接下来，举两个例子来阐释实时计算在阿里内部应用的场景：

双11大屏

每年双11阿里都会聚合有价值的数据展现给媒体，GMV大屏是其中之一。整个GMV大屏是非常典型的实时计算，每条交易数据经过聚合展现在大屏之上。从DataBase写入一条数据开始，到数据实时处理写入HBase，最后展现在大屏之上，整个过程的链路十分长。整个应用存在着许多挑战：

• 大屏展现需要秒级延迟，这需要实时计算延迟在亚秒级别
• 双11大量数据需要在一个Job中聚合完成
• Exactly-Once 保持数据计算的精确性
• 系统高可用，不存在卡顿和不可用的情况

这个应用场景的SLA非常高，要求秒级延迟和数据的精确性，但它的计算并不复杂，接下来介绍更为复杂的应用。

实时机器学习

机器学习一般有两个重要的组件：Feature 和Model。传统的机器学习使用批计算对Feature的采集和Model的训练，这样更新频率太低，无法适应数据在不断变化的应用的需求。例如在双11时，商品的价格、活动的规则与平时完全不同，依据之前的数据进行训练得不到最优的效果。因此，只有实时收集Feature并训练Model，才能拟合出较为满意的结果。为此，我们开发了实时机器学习平台。

此实时机器学习平台主要包括两个部分：实时Feature计算和实时Model计算。这套系统同样拥有很多挑战，具体如下：

• 机器学习需要采集各种各样Metrics，存在许多DataSource
• 维度多，如用户维度、商品维度。维度的叠加甚至是笛卡儿积导致最后的Metrics是海量的，State非常巨大
• 机器学习计算复杂，耗用大量CPU
• 某些数据不能存在State中，需要外部存储，存在大量外部IO

实时A/B Testing

算法工程师在调优Model时会涉及多种Model，不同的Model有不同的计算模式和方法，产生不同的计算结果。因此，往往会有不同的Query订阅实时数据，产生结果后根据用户回馈迭代Model，最终得到最优模型。A/B Tesing的挑战在于算法工程师往往计算很多Metrics，所有的Metrics都通过实时计算进行统计会浪费大量资源。

针对这个挑战，我们设计了A/B Testing的框架开发平台。它用来同步算法工程师感兴趣的Metrics进行聚合，收集起来并发送到Druid引擎。这样，算法工程师根据不同Job的要求清洗数据到Druid，最后在Druid之上对不同的Metrics进行统计分析，从而找到最优的算法Model。

总结

综上，实时计算在阿里巴巴内部存在如下挑战：

• 业务庞大，场景多，大量的机器学习需求，这些因素一起导致了计算逻辑十分复杂
• 数据量大，作业多，因此整个实时计算的机器规模十分巨大
• 要保障低延迟和数据精确性，同时要满足高吞吐量的需求

流计算的介绍

显然批量计算模型是无法满足当前大数据实时计算需求的，只有流式计算模型才是实时计算的天然计算模型，因此我先介绍下流式计算的基本思想，尤其是区别于传统批量计算的一些概念。批量计算是对于有限固定的数据集合进行处理，流式计算是对无限数据流的处理，即计算无法确定数据何时会结束。从另一个角度看，批量计算其实也可以认为是流式计算的一种特例，因此批量计算可以看做是一个数据流中的片段，即有明确开始和结束标记的数据流，如下图所示：

完善的流式计算不仅应该提供实时计算能力，还应该支持计算过程中的状态管理，状态主要是指计算过程中需要的数据或者变量，例如：统计计算中的aggregation(sum/min/max…)，机器学习中的feature和model，状态管理包括这些数据的存储、备份、恢复，版本管理，提供读写访问API，并保证一致性，如下图所示：

此外，完善的流计算还需要考虑数据的时序问题，因为现实场景中，数据的产生顺序和接收顺序未必一致，因此需要给数据附带时间戳属性，即：event time，计算逻辑可以按照数据的event time来处理，这样可以解决数据的乱序问题，配合watermark机制，可以较好的解决time window计算，如下图所示：

流和批统一的计算引擎

完整的生态系统

状态管理和一致性

Chandy-Lamport算法是Flink支持状态管理和强一致性的核心理论基础，算法基础思想如下图所示：

Chandy-Lamport算法的核心思想就是定期在流式计算任务中插入Barrier，然后触发整个流做一次Checkpoint，即将任务的State做一次Snapshot持久化保存。在下次任务重启的时候，可以基于上次成功的Checkpoint进行恢复，过程如下图所示：

Flink的选定及优化

为了应对上述挑战，我们调研了许多计算框架，最终选定Flink，原因如下：

• Flink很好地引入和设计了State，基于State复杂的逻辑计算如join能得到很好的描述
• Flink引入了Chandy-Lamport 算法，在此算法的支撑下可以完美实现Exactly-Once，并能在低延迟下实现高吞吐量。

然而，Flink在State、Chandy-Lamport 算法等方面还有很多缺陷，为此阿里开辟了名为Blink的项目。

Blink是开源Flink与阿里巴巴Improvement的结合，主要分两大块：

BlinkRuntime

包括存储、调度和计算，不同公司在使用Flink时，存储、调度以及底层优化等方面会有诸多不同，阿里巴巴的blink内部也对Runtime做了诸多个性化的优化，这一层不好与Apache Flink社区统一，我们称之为Blink Runtime。

Flink SQL

原生的Flink只有比较底层的DataStream API，用户在使用时需要设计实现大量的代码，此外DataStream本身也有设计上的缺陷。为了方便用户使用，阿里巴巴团队设计了流计算的Flink SQL并推回了社区。取名Flink SQL而不是Blink SQL，主要原因Blink和Flink在 SQL 这个用户API上面是完全和社区统一的，另外Apache Flink的大部分功能都是阿里巴巴贡献的，所以说Flink SQL就是Blink SQL，没有特别大的区别。

Blink介绍

在2015年，当时我们还是阿里巴巴搜索事业部的数据技术团队，负责阿里巴巴所有商品搜索后台的数据处理，包括淘宝，天猫，B2B等全球商品，面对海量商品的数据处理，我们需要在维护两套数据处理流程，一套是每天晚上的全量流程，同时还要一套白天的实时增量流程，为了降低开发和维护成本，我们开始探索一套流和批统一的计算引擎。

当时我们重点分析对比了Spark和Flink两套技术，最后虽然觉得Spark相对成熟稳定，但Spark是从Batch出发，模拟Streaming，而Flink正好相反是从Streaming出发，认为Batch是Streaming的Special Case，因此我们感觉Flink的设计思想更先进，更适合未来的计算发展方向，更适合我们的需求，因此我们决定选择Flink技术方向。

虽然Flink具备流计算的各种优势，但Flink在成熟度和活跃度上的不足，使得我们无法在阿里巴巴业务场景中直接使用，因此我们启动了Blink项目，目标就是扩展、优化、完善Flink，使其能够应用在阿里巴巴大规模实时计算场景，并将我们在阿里巴巴对Flink的改进都回馈给开源社区。

最近一年中Blink已经将多项架构、功能和性能改进贡献给Flink社区，例如：

• Flink架构升级，插件化原生支持不同调度系统，并实现了原生运行在Hadoop YARN上
• Failover稳定性改进，优化了Task/TaskManager以及JobManager各种组件Fail的场景处理
• 提出并实现了增量式Checkpoint的架构，使得Flink的Checkpoint/Recovery速度大幅提升，成本明显下降
• 提出并实现了Async Operator，通过异步方式，让I/O密集型计算节点的性能大幅提升
• 提出了大量Table API的全新设计，以及流和批在SQL层面的统一概念和方案

BlinkRuntime核心优化解密

部署和模型的优化

优化包含以下几点：

• 解决大规模部署问题。Flink中一个Cluster只有一个JobMaster来管理所有的Job。随着Job的不断增加，单一的Master无法承接更多的Job，产生了瓶颈。因此，我们重构了架构，使每一个Job拥有自己的Master。
• 早期的Flink中TaskManager管理很多Task，某一个Task的问题会导致TaskManager崩溃，进而影响其他Job。我们使每一个Job拥有自己的TaskManager，增强了Job的隔离。
• 引入ResourceManager。ResourceManager可以和JobMaster通讯，实时动态地调整资源，达到最优的集群部署。
• 我们不仅将这些优化应用在YarnCluster上，还应用到Mesos和Standalone的部署上。

有了这些工作，Flink就可以应用到大规模的集群部署。

Incremental Checkpoint

实时计算需要不停的在checkpoint的时候来保留计算状态。早期的Flink的checkpoint的设计存在缺陷，在每个checkpoint发生的时候，它会读取所有旧的状态数据，和新的数据合并后按照全量的方式写入磁盘。随着State的不断增大，在每次做checkpoint的时候所需要的数据读取和写入的量级是十分巨大。 这就导致Job的checkpoint的间隔需要设置的很大，不能小于1分钟。越大的checkpoint的间隔， failover的时候回退的计算就越大，造成的数据延迟也就越严重。

为了减少checkpoint间隔，我们提出了Incremental Checkpoint的设计。概括的说就是在checkpoint的时候只存储增量的state变化的数据。由于历史上每个checkpoint的数据都已经保存，后面的checkpoint只需要将不同的数据放入存储，这样每次做checkpoint需要更新的数据量就非常小，使得checkpoint可以在若干秒级内完成，这就大大减小了failover时可能引起的延迟。

异步IO

很多时候我们不得不将数据放在外部存储中，这样在计算过程中就需要通过网络IO读取数据。传统的方式使用 Sync-IO的读取方式，在发出数据请求之后，只有等待到结果返回之后才能开始下一个数据请求，这种做法造成了CPU资源的浪费，因为CPU在大多数情况下都在等待网络IO的请求返回。Sync-IO使得CPU的资源利用率无法提高到极致，也就大大影响了单位CPU下的计算吞吐。为此提升计算吞吐，我们设计了Async-IO的数据读取框架，它允许异步地多线程地读取数据。

每次数据请求发出后不需要等待数据返回就继续发送下一个数据请求。当数据请求从外部存储返回后，计算系统会调用callback方法处理数据。如果数据计算不需要保序，数据返回之后就会快速经过计算发出。如果用户需要数据的计算保序时，我们使用buffer暂时保存先到的数据，等前部数据全部到达后再批量地发送。在使用了Async-IO之后，根据设置的buffer大小不同计算吞吐可以提升几十倍甚至几百倍，这就极大地提升了单位CPU利用率和整体的计算性能。

值得一提的是，以上所述的所有Blink Runtime优化已经全部贡献给了Apache Flink社区。

Flink SQL核心功能解密

阿里完成Apache Flink SQL 80%研发工作

目前，Apache Flink SQL 80%的功能是阿里巴巴实时计算团队贡献的，包括两百个提交和近十万行代码。使用Flink SQL的原因是因为我们发现了底层API给用户的迁移、上线带来的极大不便。那么，我们又为什么选择SQL？主要原因如下：

• SQL是十分通用的描述性语言，SQL适合用来让用户十分方便的描述Job的需求。
• SQL拥有比较好的优化框架，使得用户只需要专注于业务逻辑得设计而不用关心状态管理，性能优化等等复杂得设计，这样就大大降低了使用门槛。
• SQL易懂，适合不同领域的人使用。使用SQL的用户往往都不需要特别多的计算机编程基础，从产品设计到产品开发各种人员都可以快速掌握SQL的使用方法。
• SQL的API十分稳定，在做机构升级，甚至更换计算引擎时都不用修改用户的Job而继续使用。
• 有些应用场景需要流式更新，批式验证。使用SQL可以统一批计算和流计算的查询query。真正实现一个Query，同样的结果。

流处理 VS 批处理

要想设计和批处理统一的流计算SQL，就要了解流处理和批处理的区别。两者的核心区别在于流处理的数据是无穷的而批处理的数据是有限的。这个本质区别又引入另外三个更具体的区别：

• 流处理会不断产生结果而不会结束，批处理往往只返回一个最终结果并且结束。比方说，如果要统计双11的交易金额，使用批处理计算就要在双11当天的所有交易结束后，再开始计算所有买家花费的总金额并得到一个最终数值。而流处理需要追踪实时的交易金额，实时的计算并更新结果。
• 流计算需要做checkpoint并保留状态，这样在failover的时候能够快速续跑。而批计算由于它的输入数据往往是被持久化存储过的，因此往往不需要保留状态。
• 流数据会不断更新，例如某一买家的花费总金额在不断变化，而批处理的数据是一天花费的总金额，是固定的，不会变化的。流数据处理是对最终结果的一个提前观测，往往需要把提前计算的结果撤回（Retraction）做更改而批计算则不会。

Query Configuration

上面提到的这些区别都不涉及用户的业务逻辑，也就是说这些区别不会反应在SQL的不同。我们认为这些区别只是一个job的属性不同。为了描述流计算所特有的一些属性，例如什么时候产生流计算结果和怎么保留状态，我们设计容许用户配置的Query Configuration，它主要包括两个部分：

• Latency SLA。定义了从数据产生到展现的延迟，如双11大屏是秒级别。用户根据自己的需要配置不同SLA，我们的SQL系统会根据SLA的要求做最好的优化，使得在满足用户需求的同时达到系统性能的最优。
• State Retention/TTL。流计算是永不停止的，但是流数据中的State往往不需要保留很久，保留过久势必对存储是个浪费，也极大的影响了性能。所以我们容许用户设置合理的TTL（过期时间）来获得更好的计算性能。

我们通过Query Configuration描述了流和批所不同的一些属性。接下来我们需要继续考虑如何设计流式的SQL？

动态表（Dynamic-Table）

问题关键在于SQL在批处理中对表操作而流数据中并没有表。因此，我们创建了数据会随着时间变化的动态表。动态表是流的另一种表现形式，它们之间具有对偶性，即它们可以互相转换而不破坏数据的一致性。以下是一个例子：

如图，左边是输入流，我们为每一条数据产生Dynamic-Table，再将Table的变化用Changelog发送出去。这样两次变化后，输入流和输出流中的数据始终保持一致，这就证明了引入Dynamic-Table并没有丢失语义和数据。

有了动态表的概念，我们就可以应用传统SQL作用于流上。值得一提的是，Dynamic-Table是虚拟的存在着，它并不需要实际的存储来落地。我们再来看一个例子：

如图，当有输入流的时候我们进行连续查询。我们将Stream理解为一个Dynamic-Table，动态查询是基于Dynamic-Table产生一个新的Dynamic-Table，如果需要新产生的Dynamic-Table还可以继续产生流。这里，因为加入了连续查询的聚合计算，左右两边的流已经发生了变换。总之动态表的引入提供了我们在流上做连续SQL查询的能力。

Stream SQL是没有必要存在的

通过上面的讨论，我们发现有了Dynamic-Table之后我们不需要再创造任何新的流式SQL的语义。因此我们得出这样的结论：流式SQL是没必要存在的。ANSI SQL完全可以描述Stream SQL的语义，保持ANSI SQL的标准语义是我们构建Flink SQL的一个基本原则。

ANSI SQL功能实现

基于上面的理论基础，我们继而实现了流计算所需要的若干ANSI SQL功能，包括：DML、DDL、UDF/UDTF/UDAF、连接Join、撤回(Retraction)、Window聚合等等, 除了这些功能之外，我们还做了大量的查询优化，从而保障了Flink SQL即能满足用户的各种查询的需求，同时兼具优异的查询性能。接下来，简要介绍其中几项：

1)JOIN

流和动态表具有对偶性，一条SQL看似是Table的join，事实上是流的join。

例如Inner Join的实现原理如下：数据会从输入的两边任意一条流而来，一边数据先来会被存在State中并按照Joining key查询另外一边的State，如果存在就会输出结果，不存在则不输出，直到对面数据来了之后才产生结果。

总之，两个流具有两个state，一边的数据到达后存下来等待另外一边数据，全部到达后inner join产生结果。 除了两条流的join之外，我们还引入了流和外部表的join。我们的机器学习平台会把大量的数据存储在HBase中，查询HBase中的数据的操作实际上是在连接一个外部表。连接外部表往往存在两个模式：

• Look up方式。流数据到达时即时地查询外部表，从而得到结果。
• Snapshot方式。流数据到达时即时地发送snapshot的版本信息给外部存储service从而查询数据，外部表存储根据版本信息返回结果。

值得一提的是，我们设计的这个流和外部表关联的这个功能没有引入任何新的语法，是完全按照SQL-2011的标准实现的。同样的查询在批计算上也适用。

2)Retraction

撤回是流计算的重要概念，举一个例子作解释：计算词频

词频的计算是指对所有英文单词统计频率，并最终按照频率统计不同频率下不同单词的个数。例如，如果一个统计的初始状态只有Hello World Bark三个单词，且每个单词只出现一次，那么词频的最终结果就是出现频率为1的单词有3个（出现频率为其他次数的完全没有），因此结果表只有一行“1——3”。当单词不断更新，再增加一个Hello时，因为Hello的出现频率变为2次，我们在词频的结果表中插入“2——1”这么一行新的数据。

显然，出现两次的单词是一个，那么“2——1”这个结果是对的，但是出现频率为1次的单词数已经错了，应该是2个，而不是3个。出现这种问题的本质原因是因为流计算输出的结果是对计算的一个提前观测，随着数据的不断更新，计算结果必然会发生改变，这就要求我们对之前发生的结果做撤回（retraction）再把更新的结果发出去，不然数据结果就不错误。对于上面的例子，当Hello的频率从1变到2的时候，我们不仅需要在结果表中插入“2——1”这么一行，还需要对“1——3”这一行做撤回更新操作。

值得一提的是什么时候需要撤回，什么时候不需要，完全由SQL的Query Optimizer来判断，这个用户是完全不需要感知的，用户只需要通过SQL描述他的业务计算逻辑就好了。如图所示，第一个场景不需要撤回而第二个需要，这完全是由优化框架决定而非用户 。这一点，大大体现了使用SQL，并利用SQL中所拥有的天然优化框架的好处。

3)Window聚合

Window聚合是Flink SQL的一个重要能力。图中的这个例子我们对每一个小时的数据做聚合统计。除了这种Tumble window我们还支持了Sliding Window和Session Window。将来还会支持用户自定义的window。

4)查询优化Query Optimization

除了添加新的功能，我们还做了大量的查询优化。例如micro-batching。如果没有micro-batching，处理每一条数据就会伴随着几次IO读写。有了micro-batching之后我们可以用几次IO处理来处理上千条数据。除此之外，我们还做了大量的的filter/join/aggregate pushdown以及TopN的优化，下面再举例解释TopN的优化：

如上图，我们想取销售量前三的city，对用户的Query有两种底层的实现：

• 一种方式是当没一条数据来的时候，对保存的所有city进行排序，再截取前三个city。这种设计每条数据跟新都会重新排列所有city，势必会造成大量计算资源浪费。
• 我们的Query Optimizer会自动识别到查询语句，对这种计算做优化，真正执行过程中只需要不停的更新排前三的city就可以了，这样大大优化了计算的复杂度，提升了性能。

阿里巴巴实时计算应用

阿里云流计算开发平台

一个是阿里云流计算平台（streamCompute），该平台允许用户编写SQL，并在平台内部调试debug。调试正确后，用户可以通过这个平台直接将作业发布在阿里云集群上部署，部署完成后后检测运维上线的。因此这个平台整合了所有实时计算的需求，集开发、Debug、上线部署、运维于一体，大大加速了用户开发和上线的效率。值得一提的是，2017年双11期间阿里集团绝大多数的实时计算Job均通过这个平台发布。我们今年9月开始，通过阿里云，包括公共云、专有云也将这个平台开放给外部企业，让他们能够使用到阿里巴巴实时计算的能力。

阿里实时机器学习平台Porsche

在线机器学习平台利用了Blink强大的实时计算能力，能够实时的对海量用户和商品行为数据进行流式特征提取以及训练学习，并将实时更新的特征和模型实时同步给在线的搜索和推荐引擎，实现个性化搜索和推荐，数据流程如下图所示：

为了方便算法同学开发机器学习任务，我们基于Flink SQL以及Hbase，设计实现了一个面向算法人员、支持可视化自助开发运维的在线机器学习平台——Porsche。如上图所示，用户在Porsche平台的IDE，通过可视化的方式将组件拖入画布中，配置好组件属性，定义好完整的计算DAG。这个DAG会被翻译成SQL，最终提交给Blink执行。另外，值得一提的是，Porsche平台还支持Tensorflow，今年双11也是大放异彩，本平台免去了算法同学学习使用SQL的成本，暂时只对内开放。

商品数索引构建流程

淘宝的搜索引擎是用户在淘宝购物的最主要入口，淘宝的商品数据处理和索引构建流程大致如下图所示，淘宝的商品库都存储的阿里巴巴的 MySQL 集群中，搜索的数据处理流程需要从MySQL将数据同步到搜索引擎后台的HBase存储中（类似：Google都将网页抓取到BigTable中），然后进行各种业务逻辑处理和索引构建，最终将索引推送到在线搜索引擎中提供搜索服务。

由于淘宝的商品搜索引擎需要在每天白天不断进行实时商品更新，同时晚上还需要一套额外的全量商品更新流程，因此基于Blink的统一计算模型，流式计算和批量计算可以使用一套用户逻辑代码完成。

Blink的未来

目前Blink已经在阿里巴巴内部达成共识，成为阿里巴巴统一的实时计算引擎，接下来我们将继续加大在Blink技术发展上的投入，并与开源社区更加密切的合作，突进流式计算的发展。应用场景上，一方面会继续扩大计算规模，并提推出内部统一实时计算服务，统一支持阿里内部的所有实时计算业务；另一方面也将会通过阿里云的公有云和专有云渠道向外界输出我们的实时计算能力，让更多行业和用户也都能享受到阿里巴巴实时计算的技术成果。

总之，Blink的实时计算之路刚刚开启，未来必将有更大的挑战和机遇，也非常欢迎各位对实时计算有兴趣的技术爱好者以及高校学子们投身到这件开创新一代计算平台的事情上来。

参考链接：

• https://yq.aliyun.com/articles/399401
• https://zhuanlan.zhihu.com/p/27141199