深度剖析阿里巴巴对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%以上的工作可以做到两边复用,比如一些公共的优化规则,基础数据结构等等。同时,流和批也会各自保留自己一些独特的优化和算子,以满足不同的作业行为。

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

在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的优化和改进是永无止境的,一大波改进和优化正在路上。

相关推荐