浅析背压(Back Pressure)机制及其在 Spark & Flink中的实现

源动力是丰巢内部的定期源代码剖析阅读分享会,范围不限。本次为第3期,主题是背压机制,以Spark和Flink两个目前最火的大数据项目为切入点,进行代码剖析和阅读。文字为讲述PPT的记录整理而来,会有一些口语化的痕迹。

Back Pressure是流处理系统中,非常经典而常见的问题,它是让流系统能对压力变化能够呈现良好抗压性的关键点所在。各个开源实时处理系统,在中后期,都开始有对背压机制有完善的考虑和设计,基本原理一致,但是实现方式,有依赖于具体系统而各有千秋。今天我们以Spark Streaming和Flink这两个目前最流行的流处理平台为例,剖析一下Back Pressure的原理和实现技巧。

在系统架构设计中,有一个经典的排队理论,其核心的理念是:一个服务中心的服务能力是有限的,完成服务是需要一定时间的。所以为了保障服务中心的服务能正常进行,需要在外面维护一个队列,让到达的消息事件进行排队,直到服务中心完成服务,才能让下一个事件进入服务中心。

体现这种设计理念的经典设计模式之一,就是生产者-消费者模式。生产者产生事件,消费者消费事件,而随着双方数量和处理速度的变化和不均衡,生产者生产速度超过消费者的现象,经常会存在,Queue只能够缓解这种问题,不能根治,而且Queue的长度需要被妥善的设计。

为了解决这个问题,业界提出了Reactive Stream的设计模式。这是一种 生产者-消费者 + 迭代器 的模式。它的改进点在于,消费者(Subscriber)向生产者(Publisher),指明请求的个数,然后生产者根据该数量,向订阅者推送指定数量的消息。

根据该设计模式,在Java 9中,引入标准的4个接口(Processor[Publisher, Subscriber-Subscription],并应用于RxJava。但是在大数据流式系统中,大部分只是参考其理念,而因为分布式的原因,所以需要做更加精密而复杂的设计。

在真实的生产环境中,流式系统面对的系统压力,在波峰和波谷是完全不一样的,这个时候如果用固定的资源数,会造成很大的浪费,所以在Spark和Flink中,都会有一个动态Executor个数的模型,可以动态调节。但是就算是如此,在调整的过程中,还是会出现突然的压力过大的情况,难以避免。这个时候,如何让系统能够稳健的应对压力,就需要用到背压的概念和设计了。

从上图可以看到,一个流式系统的背压能力,其实需要从输入源开始,到最后的输出,每一个位于上游的子模块/系统,都具备根据下游信号量往下游发送指定数量的消息的能力,只有这样,整个流式系统才能完美背压,不会被系统突增的压力挤垮。

Spark Streaming是Spark的流式模块,它基于Spark Core提供了一套基于micro-batch处理的实时流式处理框架。 它的基本理念是将数据流转换为DStream,再通过Spark Engine的RDD机制,进行统一处理。

上图是SparkStreaming的系统核心模块,和背压特性相关的,主要是模块3:数据的产生和导入。

基于前面的排队理论,Spark Streaming每一批次的处理时长(batch_process_time)需要小于批次间隔batch_interval,否则batch_process_time > batch_interval,程序的处理能力不足,积累的数据越来越多,最终会造成Executor的OOM。

为了避免这种情况,Spark Steaming从1.5版本开始,开始引入背压机制,第一个相关Issue是经典的SPARK-7398。其大体的思路是:

通过在Driver端进行速率估算,并将速率更新到Executor端的各个Receiver,从而实现背压

整个特性,主要由三大模块实现:

  1. 速率控制

  2. 速率估算

  3. 速率更新

速率控制

整个背压机制的核心,就是Drvier端的RateContoller,它作为控制核心,继承自StreamingListener,监听Batch的完成情况,记录下它们的关键延迟,然后传递给computeAndPublish方法,遍历Executor并进行估算和更新

速率估算

PIDRateEstimator是目前RateEstimator的唯一官方实现,基本上也没谁去重新实现一个,因为确实好用。PID(Proportional Integral Derivative,比例积分差分控制算法)是工控领域中,经过多次的验证是一种非常有效的工业控制器算法。Spark Streaming将它引入,作为根据最新的Rate,以及比例(Proportional) 积分(Integral)微分(Derivative)这3个变量,来确定最新的Rate,实现简洁明了,也非常好理解。

速率更新

计算完新Rate,就该把它发布出去了。RateController通过ReceiverTracker,利用RPC消息,发布Rate到Receiver所在的节点上,该节点上的ReceiverSupervisorImpl会接收消息,并把速率更新到BlockGenerator上,从而以控制每个批次的数据生成。

仔细阅读这两个类的代码,可以发现它们充分利用了Scala的特性和高性能网络通信库,非常的简洁,一点都不拖泥带水。无论是发送端的UpdateRateLimit的case class消息类构建,还是接收端的receive的偏函数特性,都充分的体现了Scala的代码之美。

同Spark Steaming类似,Flink的基本元素是Stream和Transformations,每个Streaming Dataflow由很多Stream和Operator组成,在这些Stream里,Source是数据源,Sink是数据池。最终会组成一个DAG。

际DAG被部署到执行集群时,还要考虑并行度的影响,假设并行度是4,同时,该集群有两个TaskManager(执行工作的节点,每个节点可以执行多个任务),假设TaskManager 1执行A.1,A.2,B.1和B.2,TaskManager 2执行A.3,A.4,B.3和B.4。于是只有不在同一台机器的子任务面临节点之间的传输,以A.1,A.2到B.3,B.4为例,如上图。

每个子任务都有自己的本地缓存池,收到的数据以及发出的数据,都会序列化之后,放入到缓冲池里。然后,两个TaskManager之间,只会建立一条物理链路(底层使用Netty通讯),所有子任务之间的通讯,都由这条链路承担。

当任何一个子任务的发送缓存(不管是子任务自己的本地缓存,还是底层传输时Netty的发送缓存)耗尽时,发送方就会被阻塞,产生背压;同样,任何任务接收数据时,如果本地缓存用完了,都会停止从底层Netty那里读取数据,这样很快上游的数据很快就会占满下游的底层接收缓存,从而背压到发送端,形成对上游所有的任务的背压。

很显然,这种思路有个明显的问题,任务一个下游子任务的产生背压,都会影响整条TaskManager之间的链路,导致全链路所有子任务背压。比如上图的B.3子任务,此时还有处理能力,但也无法收到数据。

为了解决上节的单任务背压影响全链路的问题,在Flink 1.5之后,引入了Credit-based Flow Control,基于信用点的流量控制。

这种方法,首先把每个子任务的本地缓存分为两个部分,独占缓存(Exclusive Buffers)和浮动缓存(Floating Buffers);

然后,独占缓存的大小作为信用点发给数据发送方,发送方会按照不同的子任务分别记录信用点,并发送尽可能多数据给接收方,发送后则降低对应信用点的大小;

当信用点为0时,则不再发送,起到背压的作用。在发送数据的同时,发送方还会把队列中暂存排队的数据量,发给接收方,接收方收到后,根据本地缓存的大小,决定是否去浮动缓存里请求更多的缓存来加速队列的处理,起到动态控制流量的作用。整个过程参考上图。

通过这样的设计,就实现了任务级别的背压:任意一个任务产生背压,只会影响这个任务,并不会对TaskManger上的其它任务造成影响。

可以看出,StreamTask算子处理完数据后,会调用RecordWrite将数据写到相应的ResultPartition中,每个ResultPartition会被拆分成一到多个ResultSubpartition,数据实际上,是写到每个ResultSubpartition的buffers里。如图中大图RecordWriter代码。

这里有个关键的一个方法requestNewBufferBuilder,这个方法会到Task的LocalBufferPool里请求内存, 没有就会阻塞,如上小图中代码。

Flink就是通过检测每个任务线程的栈深度来实现背压检测的,如果背压了,就会出现很深的栈深度,因为要在这个方法上等待内存释放。

CreditBasedSequenceNumberingViewReader是TaskManager对应的Netty的通讯服务中的一个部分,当调用ResultSubpartition的flush方法时,实际上调用的是CreditBasedSequenceNumberingViewReader的notifyDataAvailable方法,从而通知Netty服务端的Pipline醒来进行数据读取。

将Reader加入到准备就绪状态,在这之前会先判断reader是否可用,其实是判断Credit够不够,不够就暂时无法发出;

上图为判断Credit够不够;下图为发送数据,并减少Credit。

接收端只要及时将Credit信息发送给发送端,发送端就能根据Credit数量尽快开始数据发送。从代码结构上,和发送端相反,接收端是InputGate间接的监听Netty的读取事件:当CreditBasedPartitionRequestClientHandler正常读取出数据并写入InputChannel后,通知CheckpointBarrierHandle来读取。

只要对端可写的时候,马上尽快把Credit发送过去。

事实上,作为数据的接收方,更多是接收来自发送方的大量数据,发往发送方的Credit小相对少很多,所以一般一有Credit可用,就能马上发送给发送方,保证实时性。

Spark Streaming可以对PID算法的几个变量进行调整,以适应具体的数据流量波形;也可以调整最大处理的消息数,已防止出现OOM;Flink可以调整LocalBuffer和FloatBuffer大小,来适配不同的流量波形。

Netty传输,Flink Credit优先处理;Spark Structured Streaming采用和Flink一致的编程模型,因此也有类似的背压方法。

参考文献:

  1. Reactive Stream 介绍: http://www.infoq.com/cn/news/2015/12/reactive-streams-introduction

  2. JDK9揭秘:Reactive Streams: https://www.cnblogs.com/IcanFixIt/p/7245377.html

  3. Spark Streaming Back-Pressure Design Doc1: https://docs.google.com/document/d/1ZhiP_yBHcbjifz8nJEyPJpHqxB1FT6s8-Zk7sAfayQw/edit #

  4. Spark Streaming Back-Pressure Design Doc 2: https://docs.google.com/document/d/1ls_g5fFmfbbSTIfQQpUxH56d0f3OksF567zwA00zK9E/edit#heading=h.2mt62oi9ch9

  5. Spark Streaming back pressure: https://jaceklaskowski.gitbooks.io/spark-streaming/spark-streaming-backpressure.html

  6. Spark Streaming Backpressure - finding the optimal rate is now done automatically: https://vanwilgenburg.wordpress.com/2015/10/06/spark-streaming-backpressure/

  7. Operating on back-pressure in Spark with a PID: https://www.dropbox.com/s/dwgl7wa1z5wbkg6/PIDderivation.pdf?dl=0

  8. 深入理解Spark Streaming流量控制及反压机制 https://www.jianshu.com/p/425ba321d082

  9. A Deep-Dive into Flink's Network Stack. https://flink.apache.org/2019/06/05/flink-network-stack.html

  10. Handling Overload. https://ferd.ca/handling-overload.html

  11. https://github.com/apache/flink

  12. https://github.com/lw-lin/CoolplaySpark

我来评几句
登录后评论

已发表评论数()

相关站点

+订阅
热门文章