大沙，阿里巴巴高级技术专家，负责实时计算Flink SQL，之前在美国脸书任职，Apache Flink committer。
实时计算in阿里巴巴
1999年起，阿里从电商平台开始不断拓展业务，在金融、支付、物流、文娱各个领域衍生出众多产品，例如依托于淘宝、天猫为主的电商平台、阿里妈妈广告平台、蚂蚁金服支付宝、阿里云、大文娱等。今天的阿里它已经不仅仅是一个电商平台，而是一个庞大的应用生态。阿里巴巴目前是全球最大的电商平台，2016财年收入达到5500亿美金。在阿里平台上有5亿的用户，相当于中国人口的1/3，每天有近1000万用户通过阿里平台交易。
阿里俨然成为巨大的商业航母，在这艘航母之上，海量的用户和应用必然会产生大量的数据。目前，阿里巴巴的数据量级已经达到EB级别，每天的增长量达到PB级别，实时计算日常峰值处理的数据量可达到1亿每秒，今年双11更是达到了惊人的4.7亿每秒。
实时计算在阿里巴巴内部应用广泛。随着新经济体的出现与发展，技术的革新和用户需求的提升，人们越来越需要实时计算的能力，它的最大好处就是能够基于实时变化数据更新大数据处理的状态和结果。接下来，举两个例子来阐释实时计算在阿里内部应用的场景：
1.双11大屏
每年双11阿里都会聚合有价值的数据展现给媒体，GMV大屏是其中之一。整个GMV大屏是非常典型的实时计算，每条交易数据经过聚合展现在大屏之上。从DataBase写入一条数据开始，到数据实时处理写入HBase，最后展现在大屏之上，整个过程的链路十分长。整个应用存在着许多挑战：
1)大屏展现需要秒级延迟，这需要实时计算延迟在亚秒级别
2)双11大量数据需要在一个Job中聚合完成
3)Exactly-Once 保持数据计算的精确性
4)系统高可用，不存在卡顿和不可用的情况
这个应用场景的SLA非常高，要求秒级延迟和数据的精确性，但它的计算并不复杂，接下来介绍更为复杂的应用。
2.实时机器学习
机器学习一般有两个重要的组件：Feature 和Model。传统的机器学习使用批计算对Feature的采集和Model的训练，这样更新频率太低，无法适应数据在不断变化的应用的需求。例如在双11时，商品的价格、活动的规则与平时完全不同，依据之前的数据进行训练得不到最优的效果。因此，只有实时收集Feature并训练Model，才能拟合出较为满意的结果。为此，我们开发了实时机器学习平台。
此实时机器学习平台主要包括两个部分：实时Feature计算和实时Model计算。这套系统同样拥有很多挑战，具体如下：
1)机器学习需要采集各种各样Metrics，存在许多DataSource
2)维度多，如用户维度、商品维度。维度的叠加甚至是笛卡儿积导致最后的Metrics是海量的，State非常巨大
3)机器学习计算复杂，耗用大量CPU
4)某些数据不能存在State中，需要外部存储，存在大量外部IO
3.实时A/B Testing
用户的Query也有可能不停变化，典型的例子有实时的A/B Testing。
算法工程师在调优Model时会涉及多种Model，不同的Model有不同的计算模式和方法，产生不同的计算结果。因此，往往会有不同的Query订阅实时数据，产生结果后根据用户回馈迭代Model，最终得到最优模型。A/B Tesing的挑战在于算法工程师往往计算很多Metrics，所有的Metrics都通过实时计算进行统计会浪费大量资源。
针对这个挑战，我们设计了A/B Testing的框架开发平台。它用来同步算法工程师感兴趣的Metrics进行聚合，收集起来并发送到Druid引擎。这样，算法工程师根据不同Job的要求清洗数据到Druid，最后在Druid之上对不同的Metrics进行统计分析，从而找到最优的算法Model。
综上，实时计算在阿里巴巴内部存在如下挑战：
1)业务庞大，场景多，大量的机器学习需求，这些因素一起导致了计算逻辑十分复杂
2)数据量大，作业多，因此整个实时计算的机器规模十分巨大
3)要保障低延迟和数据精确性，同时要满足高吞吐量的需求
Flink的选定及优化
为了应对上述挑战，我们调研了许多计算框架，最终选定Flink，原因如下：
1.Flink很好地引入和设计了State，基于State复杂的逻辑计算如join能得到很好的描述
2.Flink引入了Chandy-Lamport 算法，在此算法的支撑下可以完美实现Exactly-Once，并能在低延迟下实现高吞吐量。
然而，Flink在State、Chandy-Lamport 算法等方面还有很多缺陷，为此阿里开辟了名为Blink的项目。
Blink是开源Flink与阿里巴巴Improvement的结合，主要分两大块：
1.BlinkRuntime
包括存储、调度和计算，不同公司在使用Flink时，存储、调度以及底层优化等方面会有诸多不同，阿里巴巴的blink内部也对Runtime做了诸多个性化的优化，这一层不好与Apache Flink社区统一，我们称之为Blink Runtime。
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来管理所有的Job。随着Job的不断增加，单一的Master无法承接更多的Job，产生了瓶颈。因此，我们重构了架构，使每一个Job拥有自己的Master。
2)早期的Flink中TaskManager管理很多Task，某一个Task的问题会导致TaskManager崩溃，进而影响其他Job。我们使每一个Job拥有自己的TaskManager，增强了Job的隔离。
3)引入ResourceManager。ResourceManager可以和JobMaster通讯，实时动态地调整资源，达到最优的集群部署。
4)我们不仅将这些优化应用在YarnCluster上，还应用到Mesos和Standalone的部署上。
有了这些工作，Flink就可以应用到大规模的集群部署。
2.Incremental Checkpoint
实时计算需要不停的在checkpoint的时候来保留计算状态。早期的Flink的checkpoint的设计存在缺陷，在每个checkpoint发生的时候，它会读取所有旧的状态数据，和新的数据合并后按照全量的方式写入磁盘。随着State的不断增大，在每次做checkpoint的时候所需要的数据读取和写入的量级是十分巨大。 这就导致Job的checkpoint的间隔需要设置的很大，不能小于1分钟。越大的checkpoint的间隔， failover的时候回退的计算就越大，造成的数据延迟也就越严重。
为了减少checkpoint间隔，我们提出了Incremental Checkpoint的设计。概括的说就是在checkpoint的时候只存储增量的state变化的数据。由于历史上每个checkpoint的数据都已经保存，后面的checkpoint只需要将不同的数据放入存储，这样每次做checkpoint需要更新的数据量就非常小，使得checkpoint可以在若干秒级内完成，这就大大减小了failover时可能引起的延迟。
3.异步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核心功能解密
1.阿里完成Apache Flink SQL 80%研发工作
目前，Apache Flink SQL 80%的功能是阿里巴巴实时计算团队贡献的，包括两百个提交和近十万行代码。使用Flink SQL的原因是因为我们发现了底层API给用户的迁移、上线带来的极大不便。那么，我们又为什么选择SQL？主要原因如下：
1)SQL是十分通用的描述性语言，SQL适合用来让用户十分方便的描述Job的需求。
2)SQL拥有比较好的优化框架，使得用户只需要专注于业务逻辑得设计而不用关心状态管理，性能优化等等复杂得设计，这样就大大降低了使用门槛。
3)SQL易懂，适合不同领域的人使用。使用SQL的用户往往都不需要特别多的计算机编程基础，从产品设计到产品开发各种人员都可以快速掌握SQL的使用方法。
4)SQL的API十分稳定，在做机构升级，甚至更换计算引擎时都不用修改用户的Job而继续使用。
5)有些应用场景需要流式更新，批式验证。使用SQL可以统一批计算和流计算的查询query。真正实现一个Query，同样的结果。
2.流处理 VS 批处理
要想设计和批处理统一的流计算SQL，就要了解流处理和批处理的区别。两者的核心区别在于流处理的数据是无穷的而批处理的数据是有限的。这个本质区别又引入另外三个更具体的区别：
1)流处理会不断产生结果而不会结束，批处理往往只返回一个最终结果并且结束。比方说，如果要统计双11的交易金额，使用批处理计算就要在双11当天的所有交易结束后，再开始计算所有买家花费的总金额并得到一个最终数值。而流处理需要追踪实时的交易金额，实时的计算并更新结果。
2)流计算需要做checkpoint并保留状态，这样在failover的时候能够快速续跑。而批计算由于它的输入数据往往是被持久化存储过的，因此往往不需要保留状态。
3)流数据会不断更新，例如某一买家的花费总金额在不断变化，而批处理的数据是一天花费的总金额，是固定的，不会变化的。流数据处理是对最终结果的一个提前观测，往往需要把提前计算的结果撤回（Retraction）做更改而批计算则不会。
3.Query Configuration
上面提到的这些区别都不涉及用户的业务逻辑，也就是说这些区别不会反应在SQL的不同。我们认为这些区别只是一个job的属性不同。为了描述流计算所特有的一些属性，例如什么时候产生流计算结果和怎么保留状态，我们设计容许用户配置的Query Configuration，它主要包括两个部分：
1.Latency SLA
定义了从数据产生到展现的延迟，如双11大屏是秒级别。用户根据自己的需要配置不同SLA，我们的SQL系统会根据SLA的要求做最好的优化，使得在满足用户需求的同时达到系统性能的最优。
2.State Retention/TTL
流计算是永不停止的，但是流数据中的State往往不需要保留很久，保留过久势必对存储是个浪费，也极大的影响了性能。所以我们容许用户设置合理的TTL（过期时间）来获得更好的计算性能。
我们通过Query Configuration描述了流和批所不同的一些属性。接下来我们需要继续考虑如何设计流式的SQL？
4.动态表（Dynamic-Table）
问题关键在于SQL在批处理中对表操作而流数据中并没有表。因此，我们创建了数据会随着时间变化的动态表。动态表是流的另一种表现形式，它们之间具有对偶性，即它们可以互相转换而不破坏数据的一致性。以下是一个例子：
如图，左边是输入流，我们为每一条数据产生Dynamic-Table，再将Table的变化用Changelog发送出去。这样两次变化后，输入流和输出流中的数据始终保持一致，这就证明了引入Dynamic-Table并没有丢失语义和数据。
有了动态表的概念，我们就可以应用传统SQL作用于流上。值得一提的是，Dynamic-Table是虚拟的存在着，它并不需要实际的存储来落地。我们再来看一个例子：
如图，当有输入流的时候我们进行连续查询。我们将Stream理解为一个Dynamic-Table，动态查询是基于Dynamic-Table产生一个新的Dynamic-Table，如果需要新产生的Dynamic-Table还可以继续产生流。这里，因为加入了连续查询的聚合计算，左右两边的流已经发生了变换。总之动态表的引入提供了我们在流上做连续SQL查询的能力。
5.Stream 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看似是Table的join，事实上是流的join。
例如Inner Join的实现原理如下：数据会从输入的两边任意一条流而来，一边数据先来会被存在State中并按照Joining key查询另外一边的State，如果存在就会输出结果，不存在则不输出，直到对面数据来了之后才产生结果。
总之，两个流具有两个state，一边的数据到达后存下来等待另外一边数据，全部到达后inner join产生结果。 除了两条流的join之外，我们还引入了流和外部表的join。我们的机器学习平台会把大量的数据存储在HBase中，查询HBase中的数据的操作实际上是在连接一个外部表。连接外部表往往存在两个模式：
a)Look up方式。流数据到达时即时地查询外部表，从而得到结果。
b)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有两种底层的实现：
a)一种方式是当没一条数据来的时候，对保存的所有city进行排序，再截取前三个city。这种设计每条数据跟新都会重新排列所有city，势必会造成大量计算资源浪费。
b)我们的Query Optimizer会自动识别到查询语句，对这种计算做优化，真正执行过程中只需要不停的更新排前三的city就可以了，这样大大优化了计算的复杂度，提升了性能
阿里巴巴实时计算应用
基于流计算SQL之上我们开发了两个计算平台。
1.阿里云流计算开发平台
一个是阿里云流计算平台（streamCompute），该平台允许用户编写SQL，并在平台内部调试debug。调试正确后，用户可以通过这个平台直接将作业发布在阿里云集群上部署，部署完成后后检测运维上线的。因此这个平台整合了所有实时计算的需求，集开发、Debug、上线部署、运维于一体，大大加速了用户开发和上线的效率。值得一提的是，2017年双11期间阿里集团绝大多数的实时计算Job均通过这个平台发布。我们今年9月开始，通过阿里云，包括公共云、专有云也将这个平台开放给外部企业，让他们能够使用到阿里巴巴实时计算的能力。
2.阿里实时机器学习平台Porsche
为了方便算法同学开发机器学习任务，我们基于Flink SQL以及Hbase，设计实现了一个面向算法人员、支持可视化自助开发运维的在线机器学习平台——Porsche。如上图所示，用户在Porsche平台的IDE，通过可视化的方式将组件拖入画布中，配置好组件属性，定义好完整的计算DAG。这个DAG会被翻译成SQL，最终提交给Blink执行。另外，值得一提的是，Porsche平台还支持Tensorflow，今年双11也是大放异彩，本平台免去了算法同学学习使用SQL的成本，暂时只对内开放。
双11实时计算总结
上图是阿里巴巴实时计算架构，底层是数千规模的物理机，之上是统一部署的Resource Management和Storage，然后是Blink Runtime和Flink SQL，用户通过StreamCompute和Porsche平台提交Job，现在已经在阿里内部支持了数百个工程师近千个Flink SQL Job。上述就是阿里巴巴实时计算的现状。
在实时计算的助力下，阿里双11拿到1682亿的辉煌战果，实时计算的贡献主要体现在以下几点：
1.本次双11是互联网历史最大规模的并发，每秒几十万的交易和支付的实时聚合统计操作全部是是由Blink计算带来的
2.3分01秒100亿数据的展现不仅需要较高的Data Base的高吞吐能力，还考验着实时计算的速度
3.算法平台帮助算法同学取得了很好的搜索和推荐效果，获得了整体GMV的增长
总之，实时计算不仅满足了阿里巴巴内部多种多样的需求，还提升了GMV。我们希望通过阿里云实时计算平台（StreamCompute）把Blink实时计算能力输出给阿里之外的所有企业，让他们能从中获益。
