第三章 Flink基础理论之内存优化及常见内存报错解决方案

• 哇=.=
  

1、总体内存模型

1.1、内存模型概述

​ Flink内存配置分为JobManager内存配置和TaskManager内存配置。

• JobManager：Flink集群的控制单元，本地执行模式下不需要为 JobManager 进行内存配置，配置参数将不会生效。

• TaskManager：负责执行用户代码

1.2、内存配置注意事项

①Standlone模式：独立集群需要声明分配给Flink自身内存大小，配置·Total Flink Memory较合适

②Flink On Yarn模式：容器化部署模式配置任务运行时TM占用总内存大小，超过这个值会被容器杀掉（即OOM），配置Total Process Memory

③细粒度配置方式：配置task heap (taskmanager.memory.task.heap.size) 和managed memory(taskmanager.memory.managed.size)。

• 注意事项：以上配置必须选择一种，否则Flink会启动失败；不建议同时显式地配置 Total Flink memory 和整个 Total Process Memory ，否则会冲突。
• 注意事项：Flink安装时在以下配置中选其中一种即可

# 配置一
taskmanager.memory.flink.size:1024

# 配置二
taskmanager.memory.process.size:1024

# 配置三
taskmanager.memory.task.heap.size:1024
taskmanager.memory.managed.size:1024

2、JobManager内存模型

• 内存模型图如下

2.1、JVM内存

• 含义：JVM虚拟机本身占用的内存，包括元空间和执行开销。

（1）JVM元空间

taskmanager.memory.jvm-metaspace.size: 默认256mb

（2）JVM执行开销

taskmanager.memory.jvm-overhead.fraction: 默认0.1
taskmanager.memory.jvm-overhead.min: 默认192
mbtaskmanager.memory.jvm-overhead.max: 默认1gb

注意事项：总进程内存×fraction，如果小于配置的min（或大于配置的max）大小，则使用min/max大小

2.2、Flink框架内存

• 含义：Flink框架运行时堆外内存大小，用来执行akka等外部依赖，同时也负责运行checkpoint回调及作业提交时的用户代码。

jobmanager.memory.off-heap.size: 默认128M

2.3、Job内存

• 含义：Job的作业控制内存，取决于提交的作业个数和作业的结构以及用户代码的要求

jobmanager.memory.heap.size

3、TaskManager内存模型

• 内存模型图如下

3.1、JVM内存

• 含义：JVM虚拟机本身占用的内存，包括元空间和执行开销。

（1）JVM元空间

taskmanager.memory.jvm-metaspace.size: 默认256mb

（2）JVM执行开销

taskmanager.memory.jvm-overhead.fraction: 默认0.1
taskmanager.memory.jvm-overhead.min: 默认192
mbtaskmanager.memory.jvm-overhead.max:默认1gb

注意事项：总进程内存×fraction，如果小于配置的min（或大于配置的max）大小，则使用min/max大小

3.2、框架内存

• 含义：Flink框架运行时Taskmanager本身占用的内存大小，不计入slot资源中

（1）堆内内存

taskmanager.memory.framework.heap.size:默认128MB

（2）堆外内存

taskmanager.memory.framework.off-heap.size:默认128MB

3.3、网络缓冲内存

• 含义：网络数据交换所使用的堆外内存大小，如网络数据交换缓冲区。

（1）堆外内存

堆外：taskmanager.memory.network.fraction:默认0.1
#内存区间
taskmanager.memory.network.min:默认64mb
taskmanager.memory.network.max:默认1gb

• 注意事项：Flink内存×fraction，如果小于配置的min（或大于配置的max）大小，则使用min/max大小

3.4、托管内存

• 含义：用于RocksDB State Backend 的本地内存和批的排序、哈希表、缓存中间结果。

堆外：taskmanager.memory.managed.fraction，默认0.4
#默认内存
taskmanager.memory.managed.size，默认none

• 注意事项：如果size没指定，则等于Flink内存×fraction。

3.5、Task内存

• 含义：Task执行用户代码时所使用的内存。

（1）堆内内存

taskmanager.memory.task.heap.size:默认none，由Flink内存扣除掉其他部分的内存得到。

（2）堆外内存

taskmanager.memory.task.off-heap.size:默认0，表示不使用堆外内存

4、案例分析

​ 在Flink On Yarn模式下提交任务时，一般参数指定的是总进程内存：jobmanager.memory.process.size = 4G，taskmanager.memory.process.size = 8G，则每一块得到的内存大小如下：

JobManager总内存: 4096M
TaskManager总内存: 8192M

4.1、JobManager内存分配

（1）JVM内存占用：256 + 409.6 = 665.6

• JVM元数据：256M

• JVM执行开销：4 × 1024 × 0.1 = 409.6M，在【192，1024】之间，则消耗409.6M

（2）Flink框架堆外内存：128M

（3）Job内存：4096 - 665.6 - 128 = 3302.4M

4.2、TaskManager内存分配

（1）JVM内存占用：256 + 819.2 = 1075.2M

• JVM元空间：256M
• JVM执行开销：8 × 1024 × 0.1 = 819.2M，在【192，1024】之间，则消耗819.2M

Flink内存: 7116.8M

（2）Flink框架内存：128 + 128 = 256M

• Flink框架堆外内存：128M
• Flink框架堆内内存：128M

Flink剩余内存: 6860.8M

（3）网络缓冲内存： 7116.8 × 0.1 = 711.68M

• 对外内存：7116.8M × 0.1 = 711.68M，在【192，1024】之间，则消耗711.68M

Flink剩余内存: 6,149.12M

（4）托管内存：7116.8 × 0.4 = 2,846.72M

• 托管内存：7116.8 × 0.4 = 2,846.72M

剩余内存:3,302.4M

（5）Task堆内内存：

• 堆内内存：3302.4M

Task堆内内存 = Task总内存 - (JVM内存 + Flink框架内存 + 网络缓冲内存 + 托管内存)

• 堆外内存：0M

5、任务提交资源配置（生产环境）

5.1、Flink 任务提交时内存参数配置项如下

• Standlone模式：一般配置Total Flink Memory
• Flink On Yarn模式：一般配置Total Process Memory

5.2、Standlone模式下生产环境提交参数：一般指定Flink内存

./bin/standalone-job.sh start --job-classname \
-p 5 \ 指定并行度
-Dyarn.application.queue=test \ 指定yarn队列
-Djobmanager.memory.flink.size=2048mb \ JM2~4G足够
-Dtaskmanager.memory.flink.size=4096mb \ 单个TM2~8G足够

5.3、Flink On Yarn模式下生产环境提交参数：一般指定JVM进程总体内存，可自动推导出用于Task自身的内存。

• JM内存范围：2～4G
• TM内存范围：单个2～8G

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \ 指定并行度
-Dyarn.application.queue=test \ 指定yarn队列
-Djobmanager.memory.process.size=2048mb \ JM2~4G足够
-Dtaskmanager.memory.process.size=4096mb \ 单个TM2~8G足够
-Dtaskmanager.numberOfTaskSlots=2 \ 与容器核数1core：1slot或2core：1slot

5.4、RockDB内存配置

• 注意：EmbeddedRocksDBStateBackend ：使用本机内存

默认情况RocksDB 设置为将本机内存分配限制为托管内存的大小

• 注意事项：如果禁用默认的 RocksDB 内存控制，RocksDB 分配的内存超过请求的容器大小（总进程内存）的限制，则可以在容器化部署中终止 TaskManager 。

6、Flink内存常见错误

6.1、IllegalConfigurationException

• 问题现象

TaskExecutorProcessUtils 或 JobManagerProcessUtils抛出的 IllegalConfigurationException

• 问题原因：存在无效的配置值
• 解决方案：重新配置内存参数

6.2、OutOfMemoryError: Java heap space

• 问题现象

OutOfMemoryError: Java heap space

• 问题原因：JVM Heap 太小
• 解决方案：
  • ①增加总内存来增加 JVM 堆大小
  • ② 增加TaskManager堆内存
  • ③增加JobManager的JVM 堆内存

6.3、OutOfMemoryError: Metaspace

• 问题现象

OutOfMemoryError: Metaspace 

• 问题原因： JVM 元空间限制配置得太小
• 解决方案：加总内存或者JVM元空间内存

6.4、 IOException: Insufficient number of network buffers

• 问题现象：

 IOException: Insufficient number of network buffers

• 问题原因：网络缓冲区内存不足
• 解决方案：增加总内存或网络缓冲区内存

6.5、 JobManager memory configuration failed

• 问题现象

org.apache.flink.configuration.IllegalConfigurationException: JobManager memory configuration failed: Sum of configured JVM Metaspace (256.000mb (268435456 bytes)) and JVM Overhead (192.000mb (201326592 bytes)) exceed configured Total Process Memory (256.000mb (268435456 bytes)).

• 问题原因：Flink 容器尝试分配超出其请求大小（Yarn 或 Kubernetes）的内存

• 解决方案：增加 TaskManager 的总内存。