深度剖析阿里巴巴对Apache Flink 的优化与改进

Posted 2021-04-13 极客前程

本文主要从两个层面深度剖析：阿里巴巴对Flink 究竟做了哪些优化？

取之开源，用之开源

一、 SQL 层

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

首先是对SQL 层的技术架构进行升级和替换。调研过Flink 或者使用过Flink 的开发者应该知道，深度剖析阿里巴巴对Apache Flink 的优化与改进Flink 有两套基础的API，一套是DataStream，另一套是DataSet。DataStream API 是针对流式处理的用户提供，DataSet API 是针对批处理用户提供，但是这两套API 的执行路径是完全不一样的，甚至需要生成不同的Task 去执行。Flink 原生的SQL 层在经过一系列优化之后，会根据用户希望是批处理还是流处理的不同选择，去调用DataSet 或者是DataStream API。这就会造成用户在日常开发和优化中，经常要面临两套几乎完全独立的技术栈，很多事情可能需要重复的去做

两遍。这样也会导致在一边的技术栈上做的优化，另外一边就享受不到。因此阿里巴巴在SQL层提出了全新的Quyer Processor，它主要包括一个流和批可以尽量做到复用的优化层（QueryOptimizer）以及基于相同接口的算子层（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 中写异步调用和写“HelloWord”一样简单 ，这个让Flink Job 的吞吐有了很大的飞跃。

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