The idea of setting the flash memory management and resource size, and the quantitative relationship among parallelism, slot, and taskmanager quantity

Here we use flink1.12 Of flink webUI To demonstrate memory management , Subsequent versions of the memory may have inconsistent changes , The detailed explanation mainly focuses on taskManager in

1 Memory allocation

jvm metaspace Deposit jvm Metadata of the loaded class
There are still some thoughts Network This network overhead , Half does not involve modifying memory , Not marked .

Memory data structure ：
flink Minimum memory allocation unit It's a memory segment MemorySegment Default 32k.

2 Memory setting ideas

2.1 Configuration example

stay yarn Submit to fame and fortune It can be done to yarn Configure to modify the memory size
If you use fs memory state, And there is no hash sort operation , I can modify it managed Memory Size , Let more resources be used in task In memory on the heap
-yD taskmanager.memory.size=8192 \
-yD taskmanager.memory.fraction=0.3
-yD taskmanager.memory.managed.size=0.3
Specific parameters can refer to flink Official website ：
https://nightlies.apache.org/flink/flink-docs-release-1.3/setup/config.html#managed-memory

2.2 Configuration ideas

2.2.1 Parallelism ,slot,taskmanager The relationship among quantity

To put it simply Parallelism >=slot*taskmanager ;

stay yarn Submit a flink Mission , container The quantity is calculated as follows
container.num == taskmanager.num == ( parallelism.default / taskmanager.numberOfTaskSlots )

parallelism Refer to taskmanager The concurrency capability actually used . Suppose we put parallelism.default Set up
by 1, that 9 individual TaskSlot Only use 1 individual , Yes 8 A free time . Concurrency number <=slot*tm Count slot If it is not enough, it will start automatically tm, To add

Set up parallelism There are many ways , The priority for api>env>p>file

2.2.2 Resource size selection

2.2.2.1 give an example ：

Suppose a single taskmanager by 14g,taskmanager.memory.managed.fraction by 0.5, You will get the following ：

JVM Heap Size：5.19 GB Flink Managed Memory：6.45 GB

JVM (Heap/Non-Heap) ： Heap：5.19 GB Non-Heap：1.33 GB Total：6.52 GB

Outside JVM：Capacity：1.01GB

NetWork: count: xxxxx

It can be calculated that 6.45+6.52+1.01 = 13.98 be equal to 14

2.2.2.1 The calculation process

taskmanager.memory.process.size Set the memory size of the container .
The calculation process is in org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils in processSpecFromConfig Method ,TaskExecutorProcessSpec Class shows 1.10 Version the composition of the entire memory .

The calculation method is divided into 3 Kind of ：

1. It specifies taskmanager.memory.task.heap.size and taskmanager.memory.managed.size See method ：deriveProcessSpecWithExplicitTaskAndManagedMemory
2. It specifies taskmanager.memory.flink.size See method ：deriveProcessSpecWithTotalFlinkMemory
3. It specifies taskmanager.memory.process.size（ Container environments typically specify this , Determine global capacity ）

totalProcessMemorySize = Set the value of the 14g
jvmMetaspaceSize = taskmanager.memory.jvm-metaspace.size Default 96m
The corresponding parameter -XX:MaxMetaspaceSize=100663296.

jvmOverheadSize：

• taskmanager.memory.jvm-overhead.min 192m
• taskmanager.memory.jvm-overhead.max 1g
• taskmanager.memory.jvm-overhead.fraction 0.1

The formula 14g * 0.1 = 1.4g Must be in [192m, 1g] Between , therefore jvmOverheadSize Its size is 1g

• totalFlinkMemorySize = 14g - 1g - 96m = 13216m

• frameworkHeapMemorySize：taskmanager.memory.framework.heap.size Default 128m

• frameworkOffHeapMemorySize：taskmanager.memory.framework.off-heap.size Default 128m

• taskOffHeapMemorySize：taskmanager.memory.task.off-heap.size Default 0

After determining the above parameters , It is the calculation of the three most important indicators ：taskHeapMemorySize,networkMemorySize,managedMemorySize

Calculate the score to determine ：taskmanager.memory.task.heap.size Still not sure .

1） determine taskmanager.memory.task.heap.size

• taskHeapMemorySize = Set the value

• managedMemorySize = Set the use setting value , Otherwise use 0.4 * totalFlinkMemorySize

• If taskHeapMemorySize + taskOffHeapMemorySize + frameworkHeapMemorySize + frameworkOffHeapMemorySize + managedMemorySize > totalFlinkMemorySize abnormal

• networkMemorySize Equal to the remaining size , And then check Is this memory sufficient , You can view the corresponding code by yourself

2） Not set heap size

• Make sure the managedMemorySize = Set the use setting value , Otherwise use 0.4 * totalFlinkMemorySize, Here is the 0.5 * 13216m = 6608 = 6.45g ( Here is the dashboard Display content of )

• To determine the network buffer size , There are also two cases , Don't dwell on . [64mb, 1g] 0.1 * totalFlinkMemorySize = 1321.6, So it is 1g

• The last thing left is taskHeapMemorySize, It can't be negative , This is equal to 13216 - 6608 - 1024 - 128 - 128 = 5328 = 5.2g ( This is approximately equal to dashboard Display of heap size )

3） Last jvm The calculation process of the parameters of ：

• jvmHeapSize = frameworkHeapSize + taskHeapSize = 5328 + 128 = 5456

• jvmDirectSize = frameworkOffHeapMemorySize + taskOffHeapSize + networkMemSize = 128 + 1024 = 1152

• jvmMetaspaceSize = 96m