一种基于动态水位值的Flink调度优化算法(flink1.5以前),等同于实现flink的Credit-based反压原理

优化flink反压

  • 说明
  • 1 flink反压介绍
    • 1.1 介绍
    • 1.2 大数据系统反压现状
    • 1.4 flink task与task之间的反压
    • 1.5 netty水位机制作用分析
  • 2 反压优化算法
  • 3 重点! 但是 可但是 flink1.5以后的反压过程。
  • 4 flink反压问题的查找瓶颈办法

说明

首先说明,偶然看了个论文,发现 flink优化原来比我想象中的更简单,得到了一些启发,所以写下这篇帖子,供大家共同学习。
看到的论文是《计算机科学与应用》21年11月的一篇 名字就叫做 : 一种基于动态水位值的Flink调度优化算法。感兴趣的小伙伴可以自己看一下 ,很短没多少字。
但是 很离谱的是 这篇论文中的方法在flink1.5以后的反压机制中早就实现了,我不知道 这篇论文为什么能发表在21年11月的期刊上,也可能是我看论文看的不对?反正就当学习个人实现 flink1.5的反压机制更新了

1 flink反压介绍

1.1 介绍

首先说下flink反压,其实就是flink流式处理中一种动态反馈机制,一般是在实时数据处理的过程中,上游节点的生产速度大于下游节点的消费速度,提出反馈来提醒上游,下游消费不过来了。

如垃圾回收不及时或者停顿可能使得流入系统的数据快速堆积、大促或秒杀活动时出现的流量陡增等都会造成反压。如不对反压及时处理,将会使系统资源耗尽甚至导致系统崩溃。

1.2 大数据系统反压现状

现有大数据实时处理系统处理反压问题方面,Storm 是通过监控Bolt中的接收队列负载情况,如果超过高水位值就会将反压信息写到Zookeeper,Zookeeper上的watch会通知该拓扑的所有Worker都进入反压状态,最后Spout停止发送tuple。J Storm采用逐级降速的方式来处理反压,使用Topology Master替代Zookeeper来协调拓扑进入反压状态,效果较Storm更为稳定。Spark Streaming根据批处理时间(Batch Processing Time)和批次间隔(Batch Interval,即Batch Duration)的信息来动态调整系统的摄入速率,从而完成其反压工作。

在Flink优化方面,根据论文文献目前有几种:

  1. 使用Flink执行一种传统堆序优化后的算法Heap Optimize,增加了Flink的吞吐量 。
  2. 针对Flink默认的先来先服务的任务调度策略,通过资源感知,将待执行任务分配到最佳节点进行计算,优化了Flink的负载均衡。
  3. 根据任务间数据流的大小确定拓扑边的权重,以生成关键路径,大幅缩减了Flink节点间的通信开销。
  4. 把Flink从原来的CPU迁移扩展到异构的CPU-GPU集群,在并行计算、内存管理及通信策略方面极大地提高了Flink的计算能力。

目前诸多的研究当中没有Flink反压方面的问题,当Flink面临远端传输问题时,其所依托的Netty所采用的是一种静态的水位机制,这使得Flink在面临颠簸状态数据的远程传输问题时,容易出现反复反压的情况,极大地影响了Flink传输数据的效率,故而本文将针对此问题展开研究。

例如 从我的工作经历来说,flink做简单的etl操作 ,接受kafka 数据 插入es中,数据量 大小时大时小时,实际使用来看 spark streaming 性能远远好于flink 就是因为 flink的反压

1.4 flink task与task之间的反压

Flink的反压原理如下图所示,假如Flink的一个Job分为Task A、B、C,其中Task A是Source Task、Task B处理数据、Task C为Sink Task。假如Task C由于各种原因吞吐量降低,会将负载信息反馈给Task B,Task B会降低向Task C发送数据的速率,此时若Task B还保持从Task A读取数据,数据会把Task B的Send Buffer和Receive Buffer撑爆,导致OOM或者丢失数据。所以,当Task B的Send Buffer和Receive Buffer被用完后,Task B会用同样的原理将负载信息反馈给Task A,Task A收到Task B的负载信息后,会降低给Task B发送数据的速率,以此类推。

图1

Flink反压存在Task内与跨Task两种情况,本文已在图1中标注,本文主要是针对Flink跨Task传输进行反压优化,故下文主要对Flink跨Task传输进行介绍:图2展示了Flink网络传输时的数据流向,可以看到Task Manager A给TaskManager B发送数据,Task Manager A做为Producer,Task Manager B做为Consumer。Producer端的Operator实例会产生数据,最后通过网络发送给Consumer端的Operator实例。Producer端Operator实例生产的数据首先缓存到Task Manager内部的Net Work Buffer。Net Work依赖Netty来做通信,Producer端的Netty内部有Channel Outbound Buffer,Consumer端的Netty内部有Channel Inbound Buffer。Netty最终还是要通过Socket发送网络请求,Socket这一层也会有Buffer,Producer端有Send Buffer,Consumer端有Receive Buffer。

故Flink网络传输时的整个反压过程为:首先Producer Operator从自己的上游或者外部数据源读取到数据后,对一条条的数据进行处理,处理完的数据首先输出到Producer Operator对应的Net Work Buffer中。Buffer写满或者超时后,就会触发将Net Work Buffer中的数据拷贝到Producer端Netty的Channel Outbound Buffer,之后又把数据拷贝到Socket的Send Buffer中,这里有一个从用户态拷贝到内核态的过程,最后通过Socket发送网络请求,把Send Buffer中的数据发送到Consumer端的Receive Buffer。数据到达Consumer端后,再依次从Socket的Receive Buffer拷贝到Netty的Channel Inbound Buffer,再拷贝到Consumer Operator的Net Work Buffer,最后Consumer Operator就可以读到数据进行处理了,这就是两个Task Manager之间的数据传输过程。

图2

1.5 netty水位机制作用分析

分析源码可知,Netty水位机制是一种静态的机制,Netty默认其水位线的高度为定值,这使得Flink系统在面临瞬时流量不稳定的场景(即系统的数据流量值在特别高与特别低的值之间不断跳动时)时,会出现下述两种问题:

  1. 水位值较下游可用缓存区数偏低:如图3 (左图)所示,图中以“圆圈”表示数据,以“方框”表示缓存区的大小,下同。假设当上游A点来临的数据量是9 (Flink中以buffer为数据单位,每个buffer大小为32 k,为便于表述,下文块描述),而此时下游B点的可用缓存区是10,N代表代表数据通道(其作用类似于水坝,水位值的大小决定了其单位时间通过的数据量大小),此处设水位值高度为4,则Flink传输本批次的数据需要3个单位时间(上游共9块数据,每个单位时间只能通过4块的数据,需要3个单位的时间来处理这批数据)。而若此时的水位值为9或者10的话,则只需要一个单位时间,Flink便可以处理本批次的数据。

  2. 水位值较下游可用缓存区数偏高:如图3 (右图),假设当上游A点来临的数据量是4,而此时下游B点的可用缓存区为2,水位值高度为4。由于数据量不于水位值高度,Flink会误以为可以在一单位时间内接受这批数据,如图中可以看出,只有2块的缓存区,直接接收了4块的数据量,会直接导致内存溢出(OOM)甚至引起系统阻塞。综上,由于不合理的静态水位线的设置,使得Flink传输数据时间延长,或者出现非正常的阻塞,进而影响整个Flink的数据传输情况。

图3
综上,由于不合理的静态水位线的设置,使得Flink传输数据时间延长,或者出现非正常的阻塞,进而影响整个Flink的数据传输情况。

2 反压优化算法

虽然可以在数据处理前对Netty所默认的两个buffer高度进行参数调整,但这种默认的定值始终是一种静态的机制。这种相对静态的机制使得Flink在面临远程传输问题时,容易出现上文所述的两种问题。本节将针对Flink反压传输所存在的缺点,提出一种基于动态水位值Flink调度优化算法,并给出例子进行说明。Flink-N算法的核心思想是:把Flink中Netty下游可用buffer数Bt实时写入Redis中,根据Redis中前后时刻buffer数(即Bt值)的大小变化,对水位值Wt进行动态调整,算法流程如图4所示,其具体步骤如下:

第一步,设置访问函数,并创建接口,使得Flink启动的同时运行访问函数。其中,访问函数的作用是,每间隔一段时间访问Netty下游缓存区可用buffer (图中B点位置)的数量,并将其记录到Redis中;

第二步,获得下游可用buffer数Bt;

第三步,取0.8倍的B0值的整数部分(向下取整)作为Netty的高水位值,即令W0H = ⌊0.8|B0|⌋;

第四步,将Bt值反馈到Redis中并记录;

第五步,根据Bt值调整水位值Wt,具体方法为:若Bt大于Bt与Bt−1的平均值,则Wt取Bt大于Bt与Bt−1的平均值;反之,当Bt ≤ Bt−1时,则令Wt = Bt。

第六步,重复第二步、第四步与第五步。

本文选择Redis是因为Flink处理时延是ms级别的,而Redis数据读取速度可达110,000次/s,写数据的速度可达81,000次/s,选择Redis相较于其他数据库而言,不会对Flink的时效性产生负增益。

图4
经过实验室测试,发现吞吐量有明显上升
在这里插入图片描述
cpu利用率也是

在这里插入图片描述
以及延时
在这里插入图片描述

3 重点! 但是 可但是 flink1.5以后的反压过程。

经过跟论文比较发现,实际上 flink1.5已经实现了等同于论文中设想的功能并实现。

在 Flink 层面实现反压机制,就是每一次 ResultSubPartition 向 InputChannel 发送消息的时候都会发送一个 backlog size 告诉下游准备发送多少消息,下游就会去计算有多少的 Buffer 去接收消息,算完之后如果有充足的 Buffer 就会返还给上游一个 Credit 告知他可以发送消息(图上两个 ResultSubPartition 和 InputChannel 之间是虚线是因为最终还是要通过 Netty 和 Socket 去通信)
图2

  • 1.5之前是tcp反压 链路太长 ,会涉及多个task 如果在同一个taskmanager会造成重复使用Socket 阻塞
  • 新的网路栈:TaskManager 传输数据时,不同的 TaskManager 上的两个 Subtask 间通 常根据 key 的数量有多个 Channel,这些 Channel 会复用同一个 TaskManager 级别的 TCP 链接,并且共享接收端 Subtask 级别的 Buffer Pool。
  • 在接收端,每个 Channel 在初始阶段会被分配固定数量的 Exclusive Buffer, 这些 Buffer 会被用于存储接受到的数据,交给 Operator 使用后再次被释放。 Channel 接收端空闲的 Buffer 数量称为 Credit,Credit 会被定时同步给发送端被后 者用于决定发送多少个 Buffer 的数据。
  • 在流量较大时,Channel 的 Exclusive Buffer 可能会被写满,此时 Flink 会向 Buffer Pool 申请剩余的 Floating Buffer。这些 Floating Buffer 属于备用 Buffer,哪 个 Channel 需要就去哪里。而在 Channel 发送端,一个 Subtask 所有的 Channel 会共享同一个 Buffer Pool,这边就没有区分 Exclusive Buffer 和 Floating Buffer。

看图可以知道 ,底层的通信肯定是socket,但实际用的组件是netty。是netty在互相通信,来实现反压信息的传递

在这里插入图片描述

4 flink反压问题的查找瓶颈办法

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

一般情况使用Metrics都能够定位到问题,是cpu 内存,然后再定位算子,查看设计是否有缺陷,定位节点,加Metrics
我们在监控反压时会用到的 Metrics 主要和 Channel 接受端的 Buffer 使用率有关,最为有用的是以下几个:
Metrics: Metris描述
outPoolUsage发送端 Buffer 的使用率
inPoolUsage接收端 Buffer 的使用率
floatingBuffersUsage(1.9 以上)接收端 Floating Buffer 的使用率
exclusiveBuffersUsage (1.9 以上)接收端 Exclusive Buffer 的使用率

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/694853.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

零基础学习8051单片机(十五)

本次先看书学习,并完成了课后习题,题目出自《单片机原理与接口技术》第五版—李清朝 答: (1)当 CPU正在处理某件事情的时候,外部发生的某一件事件请求 CPU 迅速去处理,于是,CPU暂时中止当前的工…

Unity之ShaderGraph如何实现水面波浪

前言 这几天通过一个水的波浪数学公式,实现了一个波浪效果,感觉成就感满满,下面给大家分享一下 首先先给大家看一下公式; 把公式转为ShaderGraph 第一行公式:waveType = z*-1*Mathf.Cos(wave.WaveAngle/360*2*Mathf.PI)+x*Mathf.Sin(WaveAngle/360*-2*Mathf.PI) 转换…

pytorch: ground truth similarity matrix

按照真实标签排序pair-wise相似度矩阵的Pytorch代码 本文仅作留档,用于输出可视化 Inputs: Ground-truths Y ∈ R n 1 \mathbf{Y}\in\mathbb R^{n\times 1} Y∈Rn1, Similarity matrix A ∈ R n n \mathbf{A}\in\mathbb R^{n\times n} A∈RnnOutputs: Block dia…

CogFixtureTool(坐标系、校正与定位)

坐标系 任何VisionPro图像都支持一组坐标空间,为表达特定特征的位置提供数字框架。最有用的空间是根空间和用户空间,根空间将点与原始获取图像中的像素相关联,用户空间用于获得校准和固定空间中的特征位置和测量值。 根空间 图像的根空间…

11、内网安全-横向移动NTLM-Relay重放Responder中继攻击LdapEws

用途:个人学习笔记,有所借鉴,欢迎指正! 目录 前提知识: 一、横向移动-NTLM 中继攻击-Relay 重放-SMB 上线 1、CS权限转给MSF: 2、MSF: 3、添加路由: 4、smb_relay重发模块: 5、受控主机输…

【算法与数据结构】797、LeetCode所有可能的路径

文章目录 一、题目二、解法三、完整代码 所有的LeetCode题解索引,可以看这篇文章——【算法和数据结构】LeetCode题解。 一、题目 二、解法 思路分析:有向无环图(Directed acyclic graph, DAG)是图论中的一个概念,它指…

2 Nacos适配达梦数据库实现方案

1、修改源代码方式 Nacos 原生是不支持达梦数据库的,所以就要想办法让它 “支持”,因为是开源软件,我们可以从源码入手,在流行的 1.x 、2.x 或最新版本代码的基本上进行修改。 主要涉及到以下内容的修改: com/alibaba/nacos/persistence/datasource/ExternalDataS

js谐音梗创意小游戏《望子成龙》

🌻 前言 龙年到来,祥瑞满天。愿您如龙般矫健,事业腾飞;如龙鳞闪耀,生活美满。祝您龙年大吉,万事如意! 龙年伊始,我给各位设计了一款原创的小游戏,话不多说,直…

智能测径仪 针对设备自身抖动都做了哪些创新加强设计

关键字:测径仪外壳设计,测径仪内部结构,外壳刚性振动,产线共振现象,镜头纯手工擦拭清洗,测径仪智能防抖算法,测径仪多重防抖技术,测径仪防抖技术,测径仪自身防抖, 在生产过程中,被测物不可避免的会发生抖动,测径仪本身也会产生抖动,只是抖动幅…

Google插件Sider: ChatGPT Sidebar + GPTs GPT-4 Turbo Sider

Sider: ChatGPT Sidebar 可以使得满屏都是机器人,左侧栏可以打开访问GPT-4. 配置跳板机地址 google 搜索的右侧也有打开

政安晨:【示例演绎机器学习】(一)—— 剖析神经网络:学习核心的Keras API

打开这篇文章,相信您已经了解了TensorFlow的一些基础知识,可以用它从头开始实现一个简单模型。 如果您对这些概念还不是太清晰,可以浏览一下我这个栏目中的相关文章: 政安晨的机器学习笔记http://t.csdnimg.cn/DHcyL 尤其是其中…

利用LaTex批量将eps转pdf、png转eps、eps转png、eps转svg

1、eps转pdf 直接使用epstopdf命令(texlive、mitex自带)。 在cmd中进入到eps矢量图片的目录,使用下面的命令: for %f in (*.eps) do epstopdf "%f" 下面是plt保存eps代码: import matplotlib.pyplot as…

网络编程知识整理

目录 1.1 引言 1.2 分层 1.3 TCP/IP的分层 1.4 互联网的地址 1.5 域名服务 1.6 封装 1.7 分用 1.8 端口号 1.1 引言 很多不同的厂家生产各种型号的计算机,它们运行完全不同的操作系统,但 T C P / I P协议族允许它们互相进行通信。这一点很让人感…

消息中间件-面试题

MQ选择 一、Kafka 1、消息队列如何保证消息可靠性 消息不重复 生产者控制消费者幂等消息不丢失 生产者发送,要确认broker收到并持久化broker确认消费者消费完,再删除消息2、kafka是什么 Kafka是一种高吞吐量、分布式、基于发布/订阅的消息中间件,是Apache的开源项目。broke…

git版本回退在eclipse和命令中的操作

一.背景 老程序员了,熟悉eclipsesvn,git用的不溜。近几年用了git,偶尔修改了某个文件希望放弃本次修改重新恢复到最新版本重新修改。或者回退到某个版本,再修改。记录一下Eclipse中的操作,和命令操作的情况。 二.Ecli…

AOSP10 替换系统launcher

本文实现将原生的launcher 移除&#xff0c;替换成我们自己写的launcher。 分以下几个步骤&#xff1a; 一、新建一个自己的launcher项目。 1.直接使用android studio 新建一个项目。 2.修改AndroidManifest.xml <applicationandroid:persistent"true"androi…

xxl-job架构原理讲解

1、调度中心 调度中心是一个单独的Web服务&#xff0c;主要是用来触发定时任务的执行 它提供了一些页面操作&#xff0c;我们可以很方便地去管理这些定时任务的触发逻辑 调度中心依赖数据库&#xff0c;所以数据都是存在数据库中的 调度中心也支持集群模式&#xff0c;但是…

mybatis数据操作语句

//基于注解 Mapper public interface reboudapt {Select("select * from dept")List<dept> huoqudept();//删除语句Delete("delete from dept where id #{id}")void deletesc(Integer id);//增加语句Insert("insert into dept(name, create_t…

Docker镜像加速

前言 众所周知&#xff0c;我们常用的一些工具或系统的下载源都是国外的&#xff0c;这就会导致我们在下载一些东西时&#xff0c;会导致下载巨慢或者下载失败的情况&#xff0c;下面便是docker换下载源的教程 镜像加速 下面是几个常用的国内的镜像 科大镜像&#xff1a;ht…

transformer,视觉模型改进论文的讨论

1、efficientVIT efficientformer 模型 快48.9倍的新SAM&#xff01;清华&MIT&英伟达开源EfficientViT-SAM&#xff1a;精度不变&#xff0c;原地起飞 YOLOv8改进 | 2023主干篇 | EfficientViT替换Backbone&#xff08;高效的视觉变换网络&#xff09; - Snu77的文章 -…