深度剖析阿里巴巴对Flink的优化与改进
实时计算团队
导读：随着人工智能时代的降临，数据量的爆发，阿里巴巴的商品数据处理就经常需要面对增量和全量两套不同的业务流程问题，所以阿里巴巴就在想：能不能有一套统一的大数据引擎技术，用户只需要根据自己的业务逻辑开发一套代码。这样在各种不同的场景下，不管是全量数据还是增量数据，亦或者实时处理，一套方案即可全部支持，这就是阿里巴巴选择 Flink 的背景和初衷。
彼时的 Flink 不管是规模还是稳定性尚未经历实践，成熟度有待商榷。阿里巴巴实时计算团队决定在阿里内部建立一个 Flink 分支 Blink，并对 Flink 进行大量的修改和完善，让其适应阿里巴巴这种超大规模的业务场景。那么，阿里巴巴对 Flink 究竟做了哪些优化呢？
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 的背景和初衷。
基于 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 究竟做了哪些优化？
取之开源，用之开源一、 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 的优化和改进是永无止境的，一大波改进和优化正在路上。
Flink 的未来方向
目前 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 作为首选的流处理引擎。
作者 | 阿里巴巴