深度剖析阿里巴巴对Apache Flink的优化与改进
本文主要从两个层面深度剖析：阿里巴巴对Flink究竟做了哪些优化？
取之开源，用之开源
一、SQL层
为了能够真正做到用户根据自己的业务逻辑开发一套代码，能够同时运行在多种不同的场景，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又大刀阔斧地进行了哪些改进。
二、Runtime层
为了让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的优化和改进是永无止境的，一大波改进和优化正在路上。