取之开源，用之开源——深度剖析阿里巴巴对Apache Flink的优化与改进

摘要：基于在阿里巴巴搭建的平台于年正式上线，并从阿里巴巴的搜索和推荐这两大场景开始实现。在经过一番调研之后，阿里巴巴实时计算认为是一个非常适合的选择。接下来，我们聊聊阿里巴巴在层对又大刀阔斧地进行了哪些改进。

Apache Flink（以下简称Flink）是诞生于欧洲的一个大数据研究项目，原名StratoSphere。该项目是柏林工业大学的一个研究性项目，早期专注于批计算。2014年，StratoSphere项目中的核心成员孵化出Flink，并在同年将Flink捐赠Apache，后来Flink顺利成为Apache的顶级大数据项目。同时Flink计算的主流方向被定位为流计算，即用流式计算来做所有大数据的计算工作，这就是Flink技术诞生的背景。

2014年Flink作为主攻流计算的大数据引擎开始在开源大数据行业内崭露头角。区别于Storm、Spark Streaming以及其他流式计算引擎的是：它不仅是一个高吞吐、低延迟的计算引擎，同时还提供很多高级功能。比如它提供有状态的计算，支持状态管理，支持强一致性的数据语义以及支持Event Time,WaterMark对消息乱序的处理等。

Flink的受欢迎还离不开它身上的众多标签，其中包括性能优秀（尤其在流计算领域）、高可扩展性、支持容错，是一种纯内存式的一个计算引擎，做了内存管理方面的大量优化，另外也支持eventime的处理、支持超大状态的Job（在阿里巴巴中作业的state大小超过TB的是非常常见的）、支持exactly-once的处理。

随着人工智能时代的降临，数据量的爆发，在典型的大数据的业务场景下数据业务最通用的做法是：选用批处理的技术处理全量数据，采用流式计算处理实时增量数据。在绝大多数的业务场景之下，用户的业务逻辑在批处理和流处理之中往往是相同的。但是，用户用于批处理和流处理的两套计算引擎是不同的。

因此，用户通常需要写两套代码。毫无疑问，这带来了一些额外的负担和成本。阿里巴巴的商品数据处理就经常需要面对增量和全量两套不同的业务流程问题，所以阿里巴巴就在想：能不能有一套统一的大数据引擎技术，用户只需要根据自己的业务逻辑开发一套代码。这样在各种不同的场景下，不管是全量数据还是增量数据，亦或者实时处理，一套方案即可全部支持，这就是阿里巴巴选择Flink的背景和初衷。

基于Flink在阿里巴巴搭建的平台于2016年正式上线，并从阿里巴巴的搜索和推荐这两大场景开始实现。目前阿里巴巴所有的业务，包括阿里巴巴所有子公司都采用了基于Flink搭建的实时计算平台。同时Flink计算平台运行在开源的Hadoop集群之上。采用Hadoop的YARN做为资源管理调度，以 HDFS作为数据存储。因此，Flink可以和开源大数据软件Hadoop无缝对接。

目前，这套基于Flink搭建的实时计算平台不仅服务于阿里巴巴集团内部，而且通过阿里云的云产品API向整个开发者生态提供基于Flink的云产品支持。

彼时的Flink不管是规模还是稳定性尚未经历实践，成熟度有待商榷。阿里巴巴实时计算团队决定在阿里内部建立一个Flink分支Blink，并对Flink进行大量的修改和完善，让其适应阿里巴巴这种超大规模的业务场景。在这个过程当中，该团队不仅对Flink在性能和稳定性上做出了很多改进和优化，同时在核心架构和功能上也进行了大量创新和改进，并将逐渐推回给社区，例如：Flink新的分布式架构，增量Checkpoint机制,基于Credit-based的网络流控机制和Streaming SQL等。接下来，我们主要从两个层面深度剖析阿里巴巴对Flink究竟做了哪些优化？

为了能够真正做到用户根据自己的业务逻辑开发一套代码，能够同时运行在多种不同的场景，Flink首先需要给用户提供一个统一的API。在经过一番调研之后，阿里巴巴实时计算认为SQL是一个非常适合的选择。在批处理领域，SQL已经经历了几十年的考验，是公认的经典。在流计算领域，近年来也不断有流表二象性、流是表的ChangeLog等理论出现。在这些理论基础之上，阿里巴巴提出了动态表的概念，使得流计算也可以像批处理一样使用SQL来描述，并且逻辑等价。这样一来，用户就可以使用SQL来描述自己的业务逻辑，相同的查询语句在执行时可以是一个批处理任务，也可以是一个高吞吐低延迟的流计算任务，甚至是先使用批处理技术进行历史数据的计算，然后自动的转成流计算任务处理最新的实时数据。在这种声明式的API之下，引擎有了更多的选择和优化空间。接下来，我们将介绍其中几个比较重要的优化。

首先是对SQL层的技术架构进行升级和替换。调研过Flink或者使用过Flink的开发者应该知道，Flink有两套基础的API，一套是DataStream，另一套是DataSet。DataStream API是针对流式处理的用户提供，DataSet API是针对批处理用户提供，但是这两套API的执行路径是完全不一样的，甚至需要生成不同的Task去执行。Flink原生的SQL层在经过一系列优化之后，会根据用户希望是批处理还是流处理的不同选择，去调用DataSet或者是DataStream API。这就会造成用户在日常开发和优化中，经常要面临两套几乎完全独立的技术栈，很多事情可能需要重复的去做两遍。这样也会导致在一边的技术栈上做的优化，另外一边就享受不到。因此阿里巴巴在SQL层提出了全新的Quyer Processor，它主要包括一个流和批可以尽量做到复用的优化层（Query Optimizer）以及基于相同接口的算子层（Query Executor）。这样一来， 80%以上的工作可以做到两边复用，比如一些公共的优化规则，基础数据结构等等。同时，流和批也会各自保留自己一些独特的优化和算子，以满足不同的作业行为。

在SQL层的技术架构统一之后，阿里巴巴开始寻求一种更高效的基础数据结构，以便让Blink在SQL层的执行更加高效。在原生Flink SQL中，都统一使用了一种叫Row的数据结构，它完全由JAVA的一些对象构成关系数据库中的一行。假如现在的一行数据由一个整型，一个浮点型以及一个字符串组成，那么Row当中就会包含一个JAVA的Integer、Double和String。众所周知，这些JAVA的对象在堆内有不少的额外开销，同时在访问这些数据的过程中也会引入不必要的装箱拆箱操作。基于这些问题，阿里巴巴提出了一种全新的数据结构BinaryRow，它和原来的Row一样也是表示一个关系数据中的一行，但与之不同的是，它完全使用二进制数据来存储这些数据。在上述例子中，三个不同类型的字段统一由JAVA的byte[]来表示。这会带来诸多好处：

首先在存储空间上，去掉了很多无谓的额外消耗，使得对象的存储更为紧凑；

其次在和网络或者状态存储打交道的时候，也可以省略掉很多不必要的序列化反序列化开销；

最后在去掉各种不必要的装箱拆箱操作之后，整个执行代码对GC也更加友好。

通过引入这样一个高效的基础数据结构，整个SQL层的执行效率得到了一倍以上的提升。

在算子的实现层面，阿里巴巴引入了更广范围的代码生成技术。得益于技术架构和基础数据结构的统一，很多代码生成技术得以达到更广范围的复用。同时由于SQL的强类型保证，用户可以预先知道算子需要处理的数据的类型，从而可以生成更有针对性更高效的执行代码。在原生Flink SQL中，只有类似a > 2或者c + d这样的简单表达式才会应用代码生成技术，在阿里巴巴优化之后，有一些算子会进行整体的代码生成，比如排序、聚合等。这使得用户可以更加灵活的去控制算子的逻辑，也可以直接将最终运行代码嵌入到类当中，去掉了昂贵的函数调用开销。一些应用代码生成技术的基础数据结构和算法，比如排序算法，基于二进制数据的HashMap等，也可以在流和批的算子之间进行共享和复用，让用户真正享受到了技术和架构的统一带来的好处。在针对批处理的某些场景进行数据结构或者算法的优化之后，流计算的性能也能够得到提升。接下来，我们聊聊阿里巴巴在Runtime层对Flink又大刀阔斧地进行了哪些改进。

为了让Flink在Alibaba的大规模生产环境中生根发芽，实时计算团队如期遇到了各种挑战，首当其冲的就是如何让Flink与其他集群管理系统进行整合。Flink原生集群管理模式尚未完善，也无法原生地使用其他其他相对成熟的集群管理系统。基于此，一系列棘手的问题接连浮现：多租户之间资源如何协调？如何动态的申请和释放资源？如何指定不同资源类型？

为了解决这个问题，实时计算团队经历大量的调研与分析，最终选择的方案是改造Flink资源调度系统，让Flink可以原生地跑在Yarn集群之上；并且重构Master架构，让一个Job对应一个Master，从此Master不再是集群瓶颈。以此为契机，阿里巴巴和社区联手推出了全新的Flip-6架构，让Flink资源管理变成可插拔的架构，为Flink的可持续发展打下了坚实的基础。如今Flink可以无缝运行在YARN、Mesos和K8s之上，正是这个架构重要性的有力说明。

解决了Flink集群大规模部署问题后，接下来的就是可靠和稳定性，为了保证Flink在生产环境中的高可用，阿里巴巴着重改善了Flink的FailOver机制。首先是Master的FailOver，Flink原生的Master FailOver会重启所有的Job，改善后Master任何FailOver都不会影响Job的正常运行；其次引入了Region-based的Task FailOver，尽量减少任何Task的FailOver对用户造成的影响。有了这些改进的保驾护航，阿里巴巴的大量业务方开始把实时计算迁移到Flink上运行。

Stateful Streaming是Flink的最大亮点，基于Chandy-Lamport算法的Checkpoint机制让Flink具备Exactly Once一致性的计算能力，但在早期Flink版本中Checkpoint的性能在大规模数据量下存在一定瓶颈，阿里巴巴也在Checkpoint上进行了大量改进，比如：

增量Checkpoint机制：阿里巴巴生产环境中遇到大JOB有几十TB
State是常事，做一次全量CP地动山摇，成本很高，因此阿里巴巴研发了增量Checkpoint机制，从此之后CP从暴风骤雨变成了细水长流；

Checkpoint小文件合并：都是规模惹的祸，随着整个集群Flink JOB越来越多，CP文件数也水涨船高，最后压的HDFS NameNode不堪重负，阿里巴巴通过把若干CP小文件合并成一个大文件的组织方式，最终把NameNode的压力减少了几十倍。

虽然说所有的数据可以放在State中，但由于一些历史的原因，用户依然有一些数据需要存放在像HBase等一些外部KV存储中，用户在Flink Job需要访问这些外部的数据，但是由于Flink一直都是单线程处理模型，导致访问外部数据的延迟成为整个系统的瓶颈，显然异步访问是解决这个问题的直接手段，但是让用户在UDF中写多线程同时还要保证ExactlyOnce语义，却并非易事。阿里巴巴在Flink中提出了AsyncOperator，让用户在Flink JOB中写异步调用和写“Hello Word”一样简单 ，这个让Flink Job的吞吐有了很大的飞跃。

Flink在设计上是一套批流统一的计算引擎，在使用过快如闪电的流计算之后，批用户也开始有兴趣入住Flink小区。但批计算也带来了新的挑战，首先在任务调度方面，阿里巴巴引入了更加灵活的调度机制，能够根据任务之间的依赖关系进行更加高效的调度；其次就是数据Shuffle，Flink原生的Shuffle Service和TM绑定，任务执行完之后要依旧保持TM无法释放资源；还有就是原有的Batch shuffle没有对文件进行合并，所以基本无法在生产中使用。阿里巴巴开发了Yarn Shuffle Service功能的同时解决了以上两个问题。在开发Yarn Shuffle Service的时候，阿里巴巴发现开发一套新的Shuffle Service非常不便，需要侵入Flink代码的很多地方，为了让其他开发者方便的扩展不同Shuffle，阿里巴巴同时改造了Flink Shuffle架构，让Flink的Shuffle变成可插拔的架构。目前阿里巴巴的搜索业务已经在使用Flink Batch Job，并且已经开始服务于生产。

经过3年多打磨，Blink已经在阿里巴巴开始茁壮生长，但是对Runtime的优化和改进是永无止境的，一大波改进和优化正在路上。

目前Flink已经是一个主流的流计算引擎，社区下一步很重要的工作是让Flink在批计算上有所突破，在更多的场景下落地，成为一种主流的批计算引擎。然后进一步在流和批之间进行无缝的切换，流和批的界限越来越模糊。用Flink,在一个计算中，既可以有流计算，又可以有批计算。

接下来阿里巴巴将致力于推动Flink在生态上得到更多语言的支持，不仅仅是Java、Scala语言，甚至是机器学习下用的Python、Go语言。

另一点不得不说AI，因为现在很多大数据计算的需求和数据量都是在支持很火爆的AI场景，所以Flink在流批生态完善的基础上，将继续完善上层的Machine Learning算法库，同时Flink也会向更成熟的机器学习、深度学习去集成。比如可以做Tensorflow On Flink, 让大数据的ETL数据处理和机器学习的Feature计算和特征计算，训练的计算等进行集成，让开发者能够同时享受到多种生态给大家带来的好处。

最后，从生态、社区的活跃来说，阿里巴巴目前在推进的一件事情是筹备2018年12月20日-21日在国家会议中心举办的首届Flink Forward China峰会（千人规模），参与者将有机会了解阿里巴巴、腾讯、华为、滴滴、美团、字节跳动等公司为何将Flink作为首选的流处理引擎。