Flink学习笔记(二):Flink内存模型

文章目录

  • 1、配置总内存
  • 2、JobManager 内存模型
  • 3、TaskManager 内存模型
  • 4、图形化展示
  • 5、实际案例计算内存分配

1、配置总内存

Flink JVM 进程的进程总内存(Total Process Memory)包含了由 Flink 应用使用的内存(Flink 总内存)以及由运行 Flink 的 JVM 使用的内存。 Flink 总内存(Total Flink Memory)包括 JVM 堆内存(Heap Memory)和堆外内存(Off-Heap Memory)。 其中堆外内存包括直接内存(Direct Memory)和本地内存(Native Memory)。详细的配置参数:https://nightlies.apache.org/flink/flink-docs-release-1.12/zh/deployment/config.html

配置 Flink 进程内存最简单的方法是指定以下两个配置项中的任意一个:

配置项TaskManager 配置参数JobManager 配置参数
Flink 总内存taskmanager.memory.flink.sizejobmanager.memory.flink.size
进程总内存taskmanager.memory.process.sizejobmanager.memory.process.size

Flink 启动需要明确配置:

TaskManagerJobManager
taskmanager.memory.flink.sizejobmanager.memory.flink.size
taskmanager.memory.process.sizejobmanager.memory.process.size
taskmanager.memory.task.heap.size 和
taskmanager.memory.managed.sizejobmanager.memory.heap.size

不建议同时设置进程总内存和 Flink 总内存。 这可能会造成内存配置冲突,从而导致部署失败。 额外配置其他内存部分时,同样需要注意可能产生的配置冲突。
JVM参数

2、JobManager 内存模型

JobManager内存模型
如上图所示,下表中列出了 Flink JobManager 内存模型的所有组成部分,以及影响其大小的相关配置参数。

组成部分配置参数描述
JVM 堆内存jobmanager.memory.heap.sizeJobManager 的 JVM 堆内存。
堆外内存jobmanager.memory.off-heap.sizeJobManager 的堆外内存(直接内存或本地内存)。
JVM Metaspacejobmanager.memory.jvm-metaspace.sizeFlink JVM 进程的 Metaspace。
JVM 开销jobmanager.memory.jvm-overhead.min、jobmanager.memory.jvm-overhead.max、jobmanager.memory.jvm-overhead.fraction用于其他 JVM 开销的本地内存,例如栈空间、垃圾回收空间等。该内存部分为基于进程总内存的受限的等比内存部分。

如配置总内存中所述,另一种配置 JobManager 内存的方式是明确指定 JVM 堆内存的大小(jobmanager.memory.heap.size)。 通过这种方式,用户可以更好地掌控用于以下用途的 JVM 堆内存大小。

3、TaskManager 内存模型

内存模型详解
如上图所示,下表中列出了 Flink TaskManager 内存模型的所有组成部分,以及影响其大小的相关配置参数。

组成部分配置参数描述
框架堆内存(Framework Heap Memory)taskmanager.memory.framework.heap.size用于 Flink 框架的 JVM 堆内存(进阶配置)。
任务堆内存(Task Heap Memory)taskmanager.memory.task.heap.size用于 Flink 应用的算子及用户代码的 JVM 堆内存。
托管内存(Managed memory)taskmanager.memory.managed.size、taskmanager.memory.managed.fraction由 Flink 管理的用于排序、哈希表、缓存中间结果及 RocksDB State Backend 的本地内存。
框架堆外内存(Framework Off-heap Memory)taskmanager.memory.framework.off-heap.size用于 Flink 框架的堆外内存(直接内存或本地内存)(进阶配置)。
任务堆外内存(Task Off-heap Memory)taskmanager.memory.task.off-heap.size用于 Flink 应用的算子及用户代码的堆外内存(直接内存或本地内存)。
网络内存(Network Memory)taskmanager.memory.network.min、taskmanager.memory.network.max、taskmanager.memory.network.fraction用于任务之间数据传输的直接内存(例如网络传输缓冲)。该内存部分为基于 Flink 总内存的受限的等比内存部分。
JVM Metaspacetaskmanager.memory.jvm-metaspace.sizeFlink JVM 进程的 Metaspace。
JVM 开销taskmanager.memory.jvm-overhead.min、taskmanager.memory.jvm-overhead.max、taskmanager.memory.jvm-overhead.fraction用于其他 JVM 开销的本地内存,例如栈空间、垃圾回收空间等。该内存部分为基于进程总内存的受限的等比内存部分。

我们可以看到,有些内存部分的大小可以直接通过一个配置参数进行设置,有些则需要根据多个参数进行调整。通常情况下,不建议对框架堆内存和框架堆外内存进行调整。 除非你非常肯定 Flink 的内部数据结构及操作需要更多的内存。 这可能与具体的部署环境及作业结构有关,例如非常高的并发度。 此外,Flink 的部分依赖(例如 Hadoop)在某些特定的情况下也可能会需要更多的直接内存或本地内存。

4、图形化展示

JobManager 内存直观展示
Job
TaskManager 内存直观展示
task
树状图表示:
树状图

5、实际案例计算内存分配

如果是 Flink On YARN 模式下:

taskmanager.memory.process.size = 4096 MB = 4G
taskmanager.memory.network.fraction = 0.15
taskmanager.memory.managed.fraction = 0.45

然后根据以上参数,就可以计算得到各部分的内存大小:

taskmanager.memory.jvm-overhead = 4096 * 0.1 = 409.6 MB
taskmanager.memory.flink.size = 4096 - 409.6 - 256 = 3430.4 MB
taskmanager.memory.network = 3430.4 * 0.15 = 514.56 MB
taskmanager.memory.managed = 3430.4 * 0.45 = 1543.68 MB
taskmanager.memory.task.heap.size = 3430.4 - 128 * 2 - 1543.68 - 514.56 = 1116.16 MB

实际案例

Flink 启动参考配置参数:

/home/dev/soft/flink/bin/flink run \-m yarn-cluster \-yD akka.ask.timeout='360 s' \-yD akka.framesize=20485760b \-yD blob.fetch.backlog=1000 \-yD blob.fetch.num-concurrent=500 \-yD blob.fetch.retries=50 \-yD blob.storage.directory=/data1/flinkdir \-yD env.java.opts.jobmanager='-XX:ErrorFile=/tmp/java_error_%p.log -XX:+PrintGCDetails -XX:-OmitStackTraceInFastThrow -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=512m -XX:+UseG1GC -XX:MaxGCPauseMillis=300 -XX:InitiatingHeapOccupancyPercent=50 -XX:+ExplicitGCInvokesConcurrent -XX:+AlwaysPreTouch -XX:AutoBoxCacheMax=20000 -XX:G1HeapWastePercent=5 -XX:G1ReservePercent=25 -Dfile.encoding=UTF-8' \-yD env.java.opts.taskmanager='-XX:ErrorFile=/tmp/java_error_%p.log -XX:+PrintGCDetails -XX:-OmitStackTraceInFastThrow -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -XX:MetaspaceSize=256m -XX:MaxMetaspaceSize=1024m -XX:+UseG1GC -XX:MaxGCPauseMillis=300 -XX:InitiatingHeapOccupancyPercent=50 -XX:+ExplicitGCInvokesConcurrent -XX:+AlwaysPreTouch -XX:AutoBoxCacheMax=20000 -Dsun.security.krb5.debug=false -Dfile.encoding=UTF-8' \-yD env.java.opts='-XX:ErrorFile=/tmp/java_error_%p.log -XX:+PrintGCDetails -XX:-OmitStackTraceInFastThrow -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -XX:MetaspaceSize=256m -XX:MaxMetaspaceSize=1024m -Dfile.encoding=UTF-8' \-yD execution.attached=false \-yD execution.buffer-timeout='1000 ms' \-yD execution.checkpointing.externalized-checkpoint-retention=RETAIN_ON_CANCELLATION \-yD execution.checkpointing.interval='30 min' \-yD execution.checkpointing.max-concurrent-checkpoints=1 \-yD execution.checkpointing.min-pause='2 min' \-yD execution.checkpointing.mode=EXACTLY_ONCE \-yD execution.checkpointing.timeout='28 min' \-yD execution.checkpointing.tolerable-failed-checkpoints=8 \-yD execution.checkpointing.unaligned=true \-yD execution.checkpointing.unaligned.forced=true \-yD heartbeat.interval=60000 \-yD heartbeat.rpc-failure-threshold=5 \-yD heartbeat.timeout=340000 \-yD io.tmp.dirs=/data1/flinkdir \-yD jobmanager.heap.size=1024m \-yD jobmanager.memory.jvm-metaspace.size=268435456b \-yD jobmanager.memory.jvm-overhead.max=1073741824b \-yD jobmanager.memory.jvm-overhead.min=1073741824b \-yD jobmanager.memory.network.fraction=0.2 \-yD jobmanager.memory.network.max=6GB \-yD jobmanager.memory.off-heap.size=134217728b \-yD jobmanager.memory.process.size='18360 mb' \-yD metrics.reporter.promgateway.deleteOnShutdown=true \-yD metrics.reporter.promgateway.factory.class=org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterFactory \-yD metrics.reporter.promgateway.filter.includes=\*:dqc\*,uptime,taskSlotsTotal,numRegisteredTaskManagers,taskSlotsAvailable,numberOfFailedCheckpoints,numRestarts,lastCheckpointDuration,Used,Max,Total,Count,Time:gauge,meter,counter,histogram \-yD metrics.reporter.promgateway.groupingKey="yarn=${yarn};hdfs=${hdfs};job_name=TEST-broadcast-${jobName//./-}-${provId}" \-yD metrics.reporter.promgateway.host=172.17.xxxx.xxxx \-yD metrics.reporter.promgateway.interval='60 SECONDS' \-yD metrics.reporter.promgateway.jobName="TEST-broadcast-${jobName//./-}-${provId}" \-yD metrics.reporter.promgateway.port=10080 \-yD metrics.reporter.promgateway.randomJobNameSuffix=true \-yD pipeline.name="TEST-broadcast-${jobName//./-}-${provId}" \-yD pipeline.object-reuse=true \-yD rest.flamegraph.enabled=true \-yD rest.server.numThreads=20 \-yD restart-strategy.failure-rate.delay='60 s' \-yD restart-strategy.failure-rate.failure-rate-interval='3 min' \-yD restart-strategy.failure-rate.max-failures-per-interval=3 \-yD restart-strategy=failure-rate \-yD security.kerberos.krb5-conf.path=/home/dev/kerberos/krb5.conf \-yD security.kerberos.login.contexts=Client,KafkaClient \-yD security.kerberos.login.keytab=/home/dev/kerberos/xxxx.keytab \-yD security.kerberos.login.principal=xxxx \-yD security.kerberos.login.use-ticket-cache=false \-yD state.backend.async=true \-yD state.backend=hashmap \-yD state.checkpoints.dir=hdfs://xxxx/flink/checkpoint/${jobName//.//}/$provId \-yD state.checkpoint-storage=filesystem \-yD state.checkpoints.num-retained=3 \-yD state.savepoints.dir=hdfs://xxxx/flink/savepoint/${jobName//.//}/$provId \-yD table.exec.hive.fallback-mapred-writer=false \-yD task.manager.memory.segment-size=4mb \-yD taskmanager.memory.framework.off-heap.size=1GB \-yD taskmanager.memory.managed.fraction=0.2 \-yD taskmanager.memory.network.fraction=0.075 \-yD taskmanager.memory.network.max=16GB \-yD taskmanager.memory.process.size='50 gb' \-yD taskmanager.network.netty.client.connectTimeoutSec=600 \-yD taskmanager.network.request-backoff.max=120000 \-yD taskmanager.network.retries=100 \-yD taskmanager.numberOfTaskSlots=10 \-yD web.timeout=900000 \-yD web.upload.dir=/data1/flinkdir \-yD yarn.application.name="TEST-broadcast-${jobName//./-}-${provId}" \-yD yarn.application.queue=$yarnQueue \-yD yarn.application-attempts=10 \

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

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

相关文章

STM32复习笔记(六):STM32远程升级BootLoader相关

目录 Preface: (一)STM32上电启动流程 (二)BootLoader相关 (三)Clion配置 Preface: 有关STM32的BootLoader主要还是参考了许多大佬的文章,这里只是简单地列举一下&am…

【FISCO-BCOS】十七、角色的权限控制

目录 一、角色定义 二、账户权限控制 1.委员新增、撤销与查询 2.委员权重修改 3.委员投票生效阈值修改 4. 运维新增、撤销与查询 一、角色定义 分为治理方、运维方、监管方和业务方。考虑到权责分离,治理方、运维方和开发方权责分离,角色互斥。 治理…

C#LINQ

LINQ(Language Integrated Query )语言集成查询,是一组用于C#和VB语言的拓展,它允许VB或者C#代码以操作内存数据的方式,查询数据库。 LINQ使用的优点: 无需复杂学习过程即可上手。编写更少代码即可创建完整应用。更快…

SpringMVC系列-4 参数解析器

背景: 本文作为SpringMVC系列的第四篇,介绍参数解析器。本文讨论的参数解析表示从HTTP消息中解析出JAVA对象或流对象并传参给Controller接口的过程。 本文内容包括介绍参数解析器工作原理、常见的参数解析器、自定义参数解析器等三部分。其中&#xff0…

【刷题篇】回溯算法(深度优先搜索(二))

文章目录 岛屿数量电话号码的字母组合组合总和活字印刷 岛屿数量 给你一个由 ‘1’(陆地)和 ‘0’(水)组成的的二维网格,请你计算网格中岛屿的数量。 岛屿总是被水包围,并且每座岛屿只能由水平方向和/或竖直…

C++笔记之不同buffer数量下的生产者-消费者机制

C笔记之不同buffer数量下的生产者-消费者机制 文章目录 C笔记之不同buffer数量下的生产者-消费者机制0.在不同的缓冲区数量下,生产者-消费者机制的实现方式和行为的区别1.最简单的生产者-消费者实现:抄自 https://mp.weixin.qq.com/s/G1lHNcbYU1lUlfugXn…

Rust中的枚举和模式匹配

专栏简介:本专栏作为Rust语言的入门级的文章,目的是为了分享关于Rust语言的编程技巧和知识。对于Rust语言,虽然历史没有C、和python历史悠远,但是它的优点可以说是非常的多,既继承了C运行速度,还拥有了Java…

Antv/s2 明细表 透视表实现和性能优化(一)

前言 以我实际项目环境为准,vuets为技术框架,代码如果有什么不懂欢迎留言评论我会回复的 透视表 定义文件 class PivotTableControl extends BaseControl {type pivotTable;label controls.chart.pivotTable;icon tc-color-pivot-table;widget () &…

目标识别项目实战:基于Yolov7-LPRNet的动态车牌目标识别算法模型(三)

前言 目标识别如今以及迭代了这么多年,普遍受大家认可和欢迎的目标识别框架就是YOLO了。按照官方描述,YOLOv8 是一个 SOTA 模型,它建立在以前 YOLO 版本的成功基础上,并引入了新的功能和改进,以进一步提升性能和灵活性…

平凡工作也能创造卓越:学习公文写作的逻辑与技巧

平凡工作也能创造卓越:学习公文写作的逻辑与技巧 前言如何把平凡的工作写出光环1.个人不能超越集体2.工作成果的概括要准确3.描写平凡工作的难点痛点 书籍介绍关键点关键词 书籍亮点内容简介购买链接参与方式往期赠书回顾 前言 如何把平凡的工作写出光环 &#x1…

docker部署Vaultwarden密码共享管理系统

Vaultwarden是一个开源的密码管理器,它是Bitwarden密码管理器的自托管版本。它提供了类似于Bitwarden的功能,允许用户安全地存储和管理密码、敏感数据和身份信息。 Vaultwarden的主要特点包括: 1. 安全的数据存储:Vaultwarden使…

运行在移动设备上的ML机器学习任务——基于MediaPipe的手势识别

前期的文章我们介绍了MediaPipe的人手关键点检测。其检测的21个点的坐标位置如下: 当检测到其关键点后,我们就可以利用此关键点来进行人手手势识别。MediaPipe 手势识别器任务可实时识别手势,并提供识别的手势结果。我们可以使用此任务来识别用户的特定手势,并调用与这些手…

Andriod 简单控件

目录 一、文本显示1.1 设置文本内容1.2 设置文本大小1.3 设置文本颜色 二、视图基础2.1 设置视图宽高2.2 设置视图间距2.3 设置视图对齐方式 三、常用布局3.1 线性布局LinearLayout3.2 相对布局RelativeLayout3.3 网格布局GridLayout3.4 滚动视图ScrollView 四、按钮触控4.1 按…

SwiftUI Spacer() onTapGesture 无法触发

问题:点击这个黑色区域不会 print,黑色区域看上去刚好是 Spacer() 占据的区域 解决办法:不使用 onTapGesture,用 Button 包裹一下 Code: import SwiftUIstruct TestTap: View {var body: some View {NavigationStack {List {Sect…

正则验证用户名和跨域postmessage

一、正则验证用户名 <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><title>登录</title> </head> <body> <form action"/login" method"post"><input type…

完美解决 flex 实现一行三个,显示多行,左对齐

效果图 代码 <body><section class"content"><div class"item">元素</div><div class"item">元素</div><div class"item">元素</div><div class"item">元素</di…

Node.js操作MySQL8.0数据库无法连接

Node.js操作MySQL8.0数据库无法连接 原创&#xff1a;丶无殇  2023-10-07 报错内容 使用node.js连接数据库MySQL 8时候&#xff0c;报错ER_NOT_SUPPORTED_AUTH_MODE&#xff0c;并且提示Client does not support authentication protocol requested by server; consider upg…

【python】可视化-绘制带有边权重的无向图

文章目录 需求示例数据代码实现 需求 输入数据表(矩阵)&#xff0c;绘制无向图。 示例数据 **示例数据1&#xff1a;**3个特征之间的关系数据 (data1.txt) featuresfeature1feature2feature3feature110.60.8feature20.610.3feature30.80.31 **示例数据2&#xff1a;**4个特…

2023年中国汽车座舱行业发展现状及趋势分析:高级人机交互(HMI)系统将逐步提升[图]

2022年有22.3%的汽车用户认为座舱内车载娱乐功能成为影响使用体验的关键因素。当前智能电动汽车的用户画像与娱乐、游戏等应用的用户画像相似&#xff0c;均以年轻人作为目标用户。年轻化的用户将娱乐功能的使用习惯延伸至汽车座舱内&#xff0c;对于座舱功能的需求不再局限于导…

C++ 33.学习C++的意义-狄泰软件学院

一些历史 UNIX操作系统诞生之初是直接用汇编语言编写的随着UNIX系统的发展&#xff0c;汇编语言的开发效率成为瓶颈&#xff0c;所以需要一个新的语言替代汇编语言1971年通过对B语言改良&#xff0c;使其能直接产生机器代码&#xff0c;C语言诞生UNIX使用C语言重写&#xff0c…