Flink任务中断：Container is running beyond physical memory limits

问题背景

某用户反馈，Flink（版本1.9）任务中断，查看日志发现用户使用的是Flink on yarn，错误日志提示如下：

Container is running beyond physical memory limits. Current usage: 99.5 GB of 99.5 GB physical memory used; 105.1 GB of 227.8 GB virtual memory used. Killing container.

为什么container占用了如此多的物理内存，从而导致任务失败呢？让我们来详细研究下。

YARN Container

在YARN上运行Flink应用程序时，两个主要的内存设定是： taskmanager.heap.size和containerized.heap-cutoff-ratio。当前环境设定如下：

taskmanager.heap.size: 102400m
containerized.heap-cutoff-ratio: 0.1

请注意，尽管taskmanager.heap.size的名称中带有单词“ heap”，但它跟JVM堆的大小没有任何关联。相反，taskmanager.heap.size才定义了从YARN请求的容器大小。

尽管我们设定了使用102400m的容器，但应用程序实际可用的内存量具有以下的关系：

total memory = taskmanager.heap.size * (1 - containerized.heap-cutoff-ratio) = 102400m * 0.9 = 92160m

容器从JVM内存中（线程堆，GC，代码等）保留cutoff了一些内存。因此，Flink任务可用的总内存为92160m。

网络内存

网络缓冲区使用的内存是从JVM off-heap内存中获取的，并且集群在flink-conf.yaml中有以下设定：

taskmanager.network.memory.max: 4gb

因此，最多为网络缓冲区分配4 GB内存。

Managed Memory

托管内存用于批处理作业，并且在集群中设定如下：

taskmanager.memory.fraction: 0.4
taskmanager.memory.off-heap: true
taskmanager.memory.preallocate: true

因此，它的内存大小可由下式计算出来：

managed memory = (total memory - network memory) * taskmanager.memory.fraction = 
  (92160m - 4096m) * 0.4 = 35225m  

JVM堆大小

现在我们可以决定为JVM堆分配多少内存。计算公式如下：

JVM heap = total memory - managed memory - network memory = 92160m - 4096m - 35225m = 52839m

Flink应用程序占用内存的主要区域如图所示：

也可以在Flink UI中查看内存的设定：

物理内存

那么，为什么container由于内存错误而被kill呢？ 当前程序启动的JVM的设定如下：

$ jps -v
18834 YarnTaskExecutorRunner -Xms52838m -Xmx52838m -XX:MaxDirectMemorySize=49562m -XX:+UseG1GC ...

可以看到Flink从一开始就声明了所有堆内存（-Xms和-Xmx相等）。

继续查看JVM堆的具体使用情况：

$ jmap -heap 18834

Garbage-First (G1) GC with 13 thread(s)

Heap Usage:
G1 Heap:
   regions  = 3303
   capacity = 55415144448 (52848.0MB)
   used     = 29174477504 (27822.94989013672MB)
   free     = 26240666944 (25025.05010986328MB)
   52.6471198345003% used
...

检查下JVM进程占用的物理内存（RES）：

$ top -p 18834

PID   USER  PR  NI  VIRT   RES    SHR    S   %CPU   %MEM   TIME+      COMMAND
18834 yarn  20  0   99.6g  97.1g  48664  S   694.0  78.1   596:25.97  java

从上面可以看到，尽管JVM堆只有52g，但已经使用了97.1g的物理内存。

更进一步查看占用内存最多的是哪一块：

$ jcmd 18834 VM.native_memory summary

Total: reserved=103154117KB, committed=102875285KB

  Java Heap (reserved=54116352KB, committed=54116352KB)

  Thread (reserved=512948KB, committed=512948KB)
             (thread #498)
             (stack: reserved=510732KB, committed=510732KB)

  GC (reserved=2151519KB, committed=2151519KB)

  Internal (reserved=45822033KB, committed=45822033KB)
...

JVM进程运行大约500个线程，每个线程需要1 MB的内存用于堆栈。还可以看到G1垃圾回收器使用了2 GB内存。

但是，最有趣的区域是“Internal”，它使用了45 GB之多！这是由直接内存缓冲区（DirectMemory）所占用的。

注意之前Flink应用程序运行时设定了

taskmanager.memory.preallocate: true 和 XX:MaxDirectMemorySize=49562m

也可以在Flink UI中看到Direct Memory的使用情况（看起来它不包括4g的网络缓冲区）：

如上可见，JVM进程的物理内存使用量与YARN容器的大小非常接近，主要的内存占用是因为直接内存缓冲区，但很小的内存峰值波动都可能迫使YARN杀死Flink Task Manager的容器，导致任务失败。

Flink 和 -XX:MaxDirectMemorySize

Flink默认使用以下公式来定义MaxDirectMemorySize的大小：

 -XX:MaxDirectMemorySize = cutoff + network memory + managed memory = 
    taskmanager.heap.size * containerized.heap-cutoff-ratio + network memory + managed memory =
    10240m + 4096m + 35225m = 49561m   

解决方案

有如下两种方案可解决“Container is running beyond physical memory limits”报错：

1. 在 yarn-site.xml中设定 yarn.nodemanager.pmem-check-enabled 为 false .

实际上，阻止YARN在分配和启动容器后检查它们使用的内存并不是一个很糟糕的决定。 可以通过使用Xmx，XX：MaxDirectMemorySize等其他限制手段来进行内存限定。

以本文为例，在具有128 GB RAM的节点上运行99.5 GB的进程是可以接受的，如果进程增加1 GB，则无需终止该进程。

1. 不要预分配managed memory，并设置taskmanager.memory.preallocate: false

这样设定可以将物理内存使用量从97.1g降到到33.9g：

PID    USER  PR   NI  VIRT    RES    SHR    S   %CPU   %MEM   TIME+     COMMAND
15534  yarn  20   0   61.3g   33.9g  48656  S   528.0  27.3   23:41.89  java

Flink UI也显示了直接内存使用量从40.9g减少到5.5g。

参考文献

1. Flink 1.9 – Off-Heap Memory on YARN – Troubleshooting Container is Running Beyond Physical Memory Limits Errors
2. Flink 原理与实现：内存管理
3. Set up Flink's Process Memory