Apache Flink provides efficient workloads on top of the JVM by tightly controlling the memory usage of its various components. While the community strives to offer sensible defaults to all configurations, the full breadth of applications that users deploy on Flink means this isn’t always possible. To provide the most production value to our users, Flink allows both high level and fine-grained tuning of memory allocation within clusters.
The further described memory configuration is applicable starting with the release version 1.10 for TaskManager and 1.11 for JobManager processes. If you upgrade Flink from earlier versions, check the migration guide because many changes were introduced with the 1.10 and 1.11 releases.
The total process memory of Flink JVM processes consists of memory consumed by the Flink application (total Flink memory) and by the JVM to run the process. The total Flink memory consumption includes usage of JVM Heap and Off-heap (Direct or Native) memory.
The simplest way to setup memory in Flink is to configure either of the two following options:
|Component||Option for TaskManager||Option for JobManager|
|Total Flink memory||
|Total process memory||
The rest of the memory components will be adjusted automatically, based on default values or additionally configured options. See also how to set up other components for TaskManager and JobManager memory.
Configuring total Flink memory is better suited for standalone deployments where you want to declare how much memory is given to Flink itself. The total Flink memory splits up into JVM Heap and Off-heap memory. See also how to configure memory for standalone deployments.
If you configure total process memory you declare how much memory in total should be assigned to the Flink JVM process. For the containerized deployments it corresponds to the size of the requested container, see also how to configure memory for containers (Kubernetes, Yarn or Mesos).
Another way to set up the memory is to configure the required internal components of the total Flink memory which are specific to the concrete Flink process. Check how to configure them for TaskManager and for JobManager.
Note One of the three mentioned ways has to be used to configure Flink’s memory (except for local execution), or the Flink startup will fail. This means that one of the following option subsets, which do not have default values, have to be configured explicitly:
|for TaskManager:||for JobManager:|
Note Explicitly configuring both total process memory and total Flink memory is not recommended. It may lead to deployment failures due to potential memory configuration conflicts. Configuring other memory components also requires caution as it can produce further configuration conflicts.
Flink explicitly adds the following memory related JVM arguments while starting its processes, based on the configured or derived memory component sizes:
|JVM Arguments||Value for TaskManager||Value for JobManager|
|-Xmx and -Xms||Framework + Task Heap Memory||JVM Heap Memory (*)|
(always added only for TaskManager, see note for JobManager)
|Framework + Task Off-heap (**) + Network Memory||Off-heap Memory (**),(***)|
|-XX:MaxMetaspaceSize||JVM Metaspace||JVM Metaspace|
(*) Keep in mind that you might not be able to use the full amount of heap memory depending on the GC algorithm used. Some GC algorithms allocate a certain amount of heap memory for themselves.
This will lead to a different maximum being returned by the Heap metrics.
(**) Notice, that the native non-direct usage of memory in user code can be also accounted for as a part of the off-heap memory.
(***) The JVM Direct memory limit is added for JobManager process only if the corresponding option
jobmanager.memory.enable-jvm-direct-memory-limit is set.
This section describes the configuration details of options which can be a fraction of some other memory size while being constrained by a min-max range:
The size of those components always has to be between its maximum and minimum value, otherwise Flink startup will fail. The maximum and minimum values have defaults or can be explicitly set by corresponding configuration options. For example, if you only set the following memory options:
then the JVM Overhead will be 1000Mb x 0.1 = 100Mb which is within the range 64-128Mb.
Notice if you configure the same maximum and minimum value it effectively fixes the size to that value.
If you do not explicitly configure the component memory, then Flink will use the fraction to calculate the memory size based on the total memory. The calculated value is capped by its corresponding min/max options. For example, if only the following memory options are set:
then the JVM Overhead will be 128Mb because the size derived from fraction is 100Mb, and it is less than the minimum.
It can also happen that the fraction is ignored if the sizes of the total memory and its other components are defined. In this case, the JVM Overhead is the rest of the total memory. The derived value still has to be within its min/max range otherwise the configuration fails. For example, suppose only the following memory options are set:
All other components of the total Process memory have default values, including the default Managed Memory fraction (or Off-heap memory in the JobManager). Then the JVM Overhead is not the fraction (1000Mb x 0.1 = 100Mb), but the rest of the total Process memory which will either be within the range 64-256Mb or fail.