只有快才不会断！阿里如何实现Blink核心技术的实时计算？

Dasha，阿里爸爸的高级技术专家，负责Flink SQL的实时计算。他之前曾在美国 Facebook 工作，是 Apache Flink 的提交者。

阿里巴巴实时数据处理

自1999年以来，阿里不断从电商平台拓展业务，依托淘宝等衍生出金融、支付、物流、娱乐等各个领域的众多产品和天猫。电商平台、阿里妈妈广告平台、蚂蚁金服支付宝、阿里云、大文娱乐等。如今的阿里不仅仅是一个电商平台，而是一个庞大的应用生态系统。阿里的巴巴目前是全球最大的电子商务平台，2016财年收入达5500亿。阿里平台有5亿用户，相当于中国人口的1/3，每天有近1000万用户通过阿里平台进行交易。

阿里已经成为一艘大型商业航母。在这艘航母上，大量的用户和应用必然会产生大量的数据。目前阿里巴巴的数据量已经达到了EB级别，日增长达到了PB级别。每日峰值数据处理实时计算可达每秒1亿次。今年的双11，每秒的浏览量达到了惊人的4.7亿。

实时数据处理在阿里巴巴内部得到广泛应用。随着新经济的出现和发展、技术的创新和用户需求的不断增加，人们越来越需要实时计算能力。其主要优点是能够根据实时变化的数据来更新大数据处理的状态和结果。接下来我们举两个例子来说明实时计算在 阿里 中的使用：

1。双11大屏

每年双11，阿里都会收集有价值的数据并向媒体展示，GMV大屏就是其中之一。一。整个GMV大屏是一个非常典型的实时计算，每笔交易数据都会被采集并显示在大屏上。从向DataBase写入一条数据，到实时处理数据并写入HBase，最后显示在大屏上，整个过程有一个很长的环节。整个应用面临很多挑战：

1）大屏显示需要秒级延迟，需要亚秒级的实时计算延迟

2）需要大量的双11数据一次作业组装

3) Exactly-Once 保持数据计算的准确性

4) 系统高可用，不存在延迟或不可用的情况

该应用场景的 SLA 很高，需要延迟以及其他级别的数据准确性，但仅此而已。计算并不复杂，接下来会介绍更复杂的应用。

2。实时机器学习

机器学习通常有两个重要组成部分：特征和模型。传统的机器学习使用批量计算来收集特征和训练模型。这个刷新率太低，无法适应数据不断变化的应用的需求。例如，双11期间，商品价格、活动规则与平时完全不同，根据过去的数据进行训练并不会达到最优效果。因此，只有实时收集特征并训练模型才能获得满意的结果。为此，我们开发了一个实时机器学习平台。

这个实时机器学习平台主要包括两部分：实时特征计算和实时模型计算。这个系统也存在很多挑战：

1）机器学习需要收集各种指标，数据源很多

2）维度很多，比如用户维度、产品维度。维度甚至笛卡尔积的叠加导致有限度量数量巨大，状态非常庞大

3）机器学习计算复杂，占用大量CPU

4）有些数据无法存储在状态并需要外部存储，有大量的外部IO

3。实时A/B测试

用户的查询也可以不断变化。一个典型的例子是实时 A/B 测试。

算法工程师在调优模型时会涉及多个模型。不同的模型有不同的计算方式和方法，给出不同的计算结果。因此，经常会有不同的查询订阅实时数据。结果生成后，根据用户反馈对模型进行迭代，最终得到最优模型。 A/B 测试的挑战在于算法工程师经常计算大量的 Metrics，通过实时计算来统计所有的 Metrics 会浪费大量的资源。

为了解决这个挑战，我们设计了一个用于A/B测试的框架开发平台。用于同步聚合算法工程师感兴趣的Metrics，收集并发送给Druid引擎。这样，算法工程师根据不同作业的需求，对Druid进行数据清洗，最后对Druid上的不同Metrics进行统计分析，找到最优的算法模型。

综上所述，实时计算在阿里巴巴面临以下挑战：

1）业务量大、场景多、机器学习需求量大。这些因素共同导致计算机逻辑非常复杂

2）数据量大，作业很多，所以整个实时计算机的规模非常巨大

3）要保证低延迟和数据准确性的同时满足高吞吐量的需求

Flink选型与优化

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

1。 Flink很好地引入和设计了状态，可以很好地描述基于状态的join等复杂逻辑计算

2。 Flink 引入了 Chandy-Lamport 算法。在该算法的支持下，可以完美实现Exactly-Once，实现低延迟的高吞吐量。

不过，Flink 在 State、Chandy-Lamport 算法等方面还存在很多不足。为此，阿里开启了一个名为Blink的项目。

Blink 是开源 Flink 和阿里巴巴改进的结合体。主要分为两部分：

1。 BlinkRuntime

包括存储、调度和数据处理。不同的公司在使用Flink时，在存储、调度、底层优化等方面会有很多差异。阿里巴巴的眨眼也对运行时做了很多个人优化。这一层并不容易与 Apache Flink 社区协调一致。我们称之为 Blink 运行时。

2。 Flink SQL

最初的 Flink 只有相对较低级别的 DataStream API。用户在使用时必须设计并实现大量的代码。另外，DataStream本身也存在设计缺陷。为了方便用户，阿里巴巴团队设计了用于流计算的Flink SQL，并将其推回社区。命名为 Flink SQL 而不是 Blink SQL 的主要原因是 Blink 和 Flink 在 SQL 用户 API 上与社区完全统一。另外，Apache Flink 的大部分功能都是阿里巴巴贡献的，所以 Flink SQL 就是 Blink SQL，没有很大的区别。

BlinkRuntime内核优化解密

1.实施和模型的优化

优化包括以下几点：

1）解决大规模的实施问题。在Flink中，一个Cluster只有一个JobMaster来管理所有作业。随着作业数量不断增加，单个Master无法处理更多作业，形成瓶颈。因此，我们重构了架构，让每个作业都有自己的Master。

2）在 Flink 早期，TaskManager 处理了很多任务。特定任务的问题会导致 TaskManager 崩溃，从而影响其他作业。我们使每个作业都有自己的 TaskManager，从而增强作业隔离性。

3) 引入ResourceManager。 ResourceManager可以与JobMaster通信，实时动态调整资源，实现最优的集群部署。

4) 我们不仅将这些优化应用于 YarnCluster，还应用于 Mesos 和 Standalone 实现。

通过这些任务，Flink 可以用于大规模集群部署。

2.增量检查点

实时计算需要不断的检查点来维持计算状态。早期 Flink 检查点的设计存在缺陷。当每个检查点发生时，它将读取所有旧状态数据，将其与新数据合并，并将其完整写入磁盘。随着状态不断增长，每次执行检查点所需的数据读写量非常大。这意味着作业的检查点间隔设置得很大，不能小于1分钟。检查点间隔越大，故障转移时的回滚计算量越大，数据延迟就越严重。

为了减少Checkpoint间隔，我们提出了Incremental Checkpoint的设计。总而言之，在检查点期间仅存储增量状态更改数据。由于历史上每个检查点的数据都已保存，后续检查点只需存储不同的数据即可。这样，每个检查点需要更新的数据量就很小，从而可以在几秒内完成检查点。这大大减少了故障转移期间可能发生的延迟。

3。异步IO

很多时候我们需要将数据放到外部存储中，所以在计算过程中必须通过网络IO读取数据。传统方法采用Sync-IO读取方法。发出一次数据查询后，必须等待结果返回后才能开始下一次数据查询。这种做法会造成CPU资源的浪费，因为大多数情况下CPU都是这样。等待网络IO请求返回。Sync-IO 可以防止 CPU 资源利用率上升到极限，从而极大地影响每台计算机的计算吞吐量CPU单元。为了提高计算吞吐量，我们设计了Async-IO数据读取框架，允许异步多线程读取数据。

每发送一个数据请求，无需等待数据返回才继续发送下一个数据请求。当数据请求从外部存储返回时，计算机系统将调用回调方法来处理数据。如果数据计算不需要订单存储，则返回后会快速计算数据并发送出去。如果用户需要按顺序计算数据，我们用一个缓冲区来临时存储先到达的数据，等前面的数据全部到达后再分批发送。使用Async-IO后，根据设置的缓冲区大小，计算吞吐量可以提高数十倍甚至数百倍，大大提高设备的CPU利用率和整体计算性能。

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

解密Flink SQL核心函数

1. 阿里 完成了 Apache Flink SQL 80% 的研发工作

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

1）SQL是一种非常常见的描述性语言，SQL适合用户非常方便地描述作业需求。

2）SQL有比较好的优化框架，用户只需要专注于业务逻辑设计，而不用担心状态管理、性能优化等复杂的设计，大大降低了使用门槛。

3）SQL简单易懂，适合各个领域的人。经常使用 SQL 的用户不需要有太多的计算机编程背景。从产品设计到产品开发的所有人员都可以快速掌握SQL的使用。

4) SQL API 非常稳定。当你升级组织甚至改变计算引擎时，你可以继续使用它，而无需改变用户的工作。

5）部分应用场景需要流式更新和批量验证。 SQL可用于统一批处理和流计算的查询。真正实现一个查询，结果是一样的。

2。流处理 VS 批处理

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

1）流处理会不断地产生结果，而批处理通常只返回最终结果并终止。例如，如果您想使用批量计算的方式来统计双11的交易金额，则开始计算所有买家的消费总额，并在双11结束时得到所有交易后的最终值。流处理必须实时跟踪交易金额，实时计算和更新结果。

2）流计算必须控制和维护状态，以便在发生故障转移时能够快速继续运行。在批处理计算中，通常不需要保留状态，因为输入数据通常是持久存储的。

3）流媒体数据将持续更新。例如，某个买家的消费总额是不断变化的，而批量数据是一天的消费总额，是固定的，不会变化。流式计算是对最终结果的预观察，往往需要将预计算的结果撤回（回滚）进行更改，而批量计算则不需要。

3.查询配置

以上差异均不涉及用户的业务逻辑，也就是说这些差异不会体现在SQL的差异上。我们认为，这些差异只是工作的不同特征。为了描述流计算特有的一些属性，例如何时生成流计算结果以及如何保存状态，我们设计了允许用户配置的查询配置。主要由两部分组成：

1。延迟SLA

定义了从数据产生到呈现的延迟，比如双11大屏，是另一个层面的。用户根据自己的需求配置不同的SLA。我们的SQL系统会根据SLA的要求进行最佳的优化，以达到最佳的系统性能，同时满足用户的需求。

2.状态保留/TTL

流计算永远不会停止，但流数据的状态往往不需要长期存储。保持太久必然会浪费存储空间并且极大影响性能。因此，我们允许用户设置合理的TTL（过期时间），以达到更好的计算性能。

我们通过查询配置描述了流式和批式之间不同的一些属性。接下来我们要继续考虑如何设计流式SQL？

4。 Dynamic-Table

问题的关键是批处理时SQL是对表进行操作的，而流数据中是没有表的。这就是为什么我们创建数据随时间变化的动态表。动态表是流的另一种表示形式。它们具有对偶性，即可以相互转换而不破坏数据的一致性。下面是一个例子：

如图所示，左侧是输入流。我们为每条数据生成一个动态表，然后使用 Changelog 将更改推送到表中。经过这两次改变，输入流和输出流中的数据始终保持一致，这证明Dynamic-Table的引入并没有丢失语义和数据。

有了动态表的概念，我们就可以将传统的SQL应用到流中。值得一提的是，Dynamic-Table是虚拟存在的，不需要实际存储来实现。我们再看一个例子：

如图所示，当有输入流时，我们进行连续查询。我们将Stream理解为一个动态表。动态查询根据动态表生成新的动态表。如果需要新生成的Dynamic-Table，则可以继续生成流。这里由于增加了连续查询的聚合计算，左右流已经进行了转换。简而言之，动态表的引入使我们能够对流进行连续的SQL查询。

5。流式 SQL 不需要存在

通过上面的讨论，我们发现使用 Dynamic-Table 我们不需要创建任何新的流式 SQL 语义。因此，我们得出结论：流式SQL不需要存在。 ANSI SQL可以完整描述Stream SQL的语义。维护 ANSI SQL 的标准语义是我们构建 Flink SQL 的基本原则。

6。 ANSI SQL函数实现

基于以上理论基础，我们接下来实现了流计算所需的几个ANSI SQL函数，包括：DML、DDL、UDF/UDTF/UDAF、Join、Retraction）、Window聚合等。除了这些特性之外，我们还做了很多查询优化，以确保 Flink SQL 能够满足用户各种查询需求的同时拥有优异的查询性能。接下来简单介绍一下其中的几个：

1) JOIN

流和动态表具有二元性。 SQL看似是表连接，但实际上它是流连接。

例如Inner Join的实现原理如下：数据会来自输入流的两侧。一侧的数据会先存入状态，然后根据连接键查询另一侧的状态。如果找到，则输出结果。如果找到，则不会读取，直到相反的数据到达后才会生成结果。

简单来说，两个流有两种状态。一侧的数据到达后，存储起来等待另一侧的数据。当所有数据到达时，内连接产生结果。除了两个流的join之外，我们还介绍了流和外部表的join。我们的机器学习平台会在HBase中存储大量的数据。查询HBase中数据的操作实际上是连接外部表。连接外部表通常有两种模式：

a) 查找模式。当流数据到达时立即查询外部表以获得结果。

b)快照方法。当流式数据到达时，立即将快照版本信息发送到外部存储服务查询数据，外部表存储根据版本信息返回结果。

值得一提的是，我们设计的流映射到外部表的函数没有引入任何新的语法，完全按照SQL-2011标准实现。相同的查询也适用于批量计算。

2）Retraction

Retraction是流计算中的一个重要概念。这里举个例子来说明一下： 计算词频

计算词频是指统计所有英文单词出现的频率，最后根据频率数统计不同频率的不同单词。例如，如果统计的初始状态只有三个词Hello World Bark，并且每个词只出现一次，那么词频的最终结果是有3个词出现频率为1（没有其他词出现的词）发生频率）。因此结果表只有一行“1-3”。当单词不断更新，又新增一个Hello时，由于Hello的出现频率变成了2次，所以我们在词频结果表中插入“2-1”等新的一行数据。

显然，出现两次的单词是1，所以结果“2-1”是正确的，但出现一次的单词数量是错误的。必须是2，而不是3。这个问题的主要原因是当前计算的结果是计算的事先观察。随着数据的不断更新，计算结果不可避免地会发生变化，这就需要我们撤回之前的结果。然后发送更新后的结果，不然数据结果就不会错了。对于上面的例子，当Hello的频率从1变为2时，我们不仅要向结果表中插入“2-1”行，还要撤消对“1-3”行的更新操作。

值得一提的是，什么时候应该收回，什么时候不应该完全由SQL查询优化器决定。这个用户根本不需要感知它。用户只需要通过SQL描述自己的业务计算逻辑。 。如图所示，第一种场景不需要撤回，而第二种场景需要撤回，这完全是由优化框架决定的，而不是由用户决定的。这很大程度上体现了使用SQL以及利用SQL中包含的自然优化框架的好处。

3）窗口聚合

窗口聚合是Flink SQL的一个重要特性。在图中的示例中，我们对每小时的数据创建聚合统计。除了这个翻滚窗口之外，我们还支持滑动窗口和会话窗口。将来还将支持自定义窗口。

4）查询优化

除了添加新功能之外，我们还做了很多查询优化。例如，微批处理。如果没有微批处理，每条数据的处理都会伴随着无数的IO读写。通过微批处理，我们可以通过多个 IO 进程处理数千条数据。另外，我们还做了一些filter/join/aggregate下推和TopN优化。下面举个例子来解释一下TopN的优化：

如上图所示，我们想取销量排名前三的城市，对用户的影响查询有两个底层实现：

a ）一种方法是在没有数据到来时对所有保存的城市进行排序，然后截取前三个城市。这种设计每次更新一条数据都会重新排列所有城市，这必然会造成大量计算资源的浪费。

b) 我们的查询优化器会自动识别查询短语并优化此计算。在实际执行过程中，我们只需要不断更新前三名的城市即可，这样就大大优化了计算的复杂度。 ，提高性能

阿里巴巴实时计算应用

我们开发了两个基于流计算SQL的计算平台。

1。 阿里云流计算开发平台

一是阿里云流计算平台（streamCompute），允许用户在平台上编写SQL和调试。一旦调试正确，用户可以通过该平台将作业直接发布到阿里云集群进行部署。实施完成后，运维将受控、上线。因此，该平台实时集成了所有数据处理需求，集开发、调试、上线实施和运维于一体，大大加快了用户开发和上线的效率。值得一提的是，2017年双11期间，阿里集团的大部分实时电脑作业都是通过这个平台发布的。从今年9月份开始，我们还将通过阿里云，包括公有云和私有云，向外部企业开放这个平台，让他们能够使用阿里爸爸的实时计算能力。

2。实时保时捷机器学习平台保时捷

为了方便算法同学开发机器学习任务，我们设计并实现了面向算法人员的在线机器学习平台，支持基于Flink SQL和Hbase的可视化自助开发和运营 — —保时捷。如上图所示，用户可视化地将组件拖入保时捷平台IDE的画布中，配置组件属性并定义完整的计算DAG。这个DAG会被翻译成SQL，最后发送到Blink执行。另外，值得一提的是，保时捷平台还支持Tensorflow，今年双11期间也大放异彩。该平台节省了算法学生学习如何使用SQL的成本，目前仅对内部用户开放。

双11实时计算总结

上图是阿里爸爸的实时计算架构。最底层是数千台物理机。上面是资源管理和存储的统一部署，然后是Blink Runtime和Flink SQL。用户可以通过StreamCompute和Porsche平台提交作业，现在支持阿里数百名工程师的近千个Flink SQL作业。以上是阿里巴巴实时数据处理的现状。

利用实时计算，阿里在双11取得了1682亿的辉煌成绩。实时计算的贡献主要体现在以下几点：

1。本次双11是互联网史上规模最大的同步活动。每秒数十万笔交易、支付的实时统计聚合运算，全部由 Blink 计算产生

2.3 分 1 秒百亿条数据的展现，不仅需要数据库的高吞吐能力，还测试实时计算速度

3。算法平台帮助算法同学取得了良好的搜索和推荐结果，实现了整体GMV的增长

总之，实时计算不仅满足了阿里巴巴的各种需求，还提高了GMV。我们希望通过阿里云实时计算平台（StreamCompute）将Blink的实时计算能力输出到阿里以外的所有企业，让他们从中受益。