I am measuring memory usage for an application (WordCount) in Flink with ps -p TaskManagerPID -o rss. However the results don’t make any sense. Because for every amount of data (1MB, 10MB, 100MB, 1GB, 10GB) there is the same amount of memory used. For 10GB data the result of the measurement is even less than 10GB. Is TaskManager the wrong process for measuring memory usage? Which process of the Flink Process Model is responsible for memory allocation?
Flink features two processing modes, stream and batch processing.
Stream Processing:
In stream processing, Flink uses pluggable state backends to maintain the state of an applications. In Flink version 1.5.0, there are two types of state backends. 1) backends (FsStateBackend and MemoryStateBackend) that store the application state on the heap of the worker (TaskManager) JVM process and 2) the RocksDBStateBackend that stores the state in RocksDB on disk. In both cases, you can monitor the memory consumption using regular JVM memory monitoring tools. However, for the RocksDBStateBackend most of the state will be stored on disk.
Batch Processing
The internal processing algorithms (sorting, hash tables) of the batch processing operators work with managed memory that is (typically) pre-allocated when a worker processes (TaskManager) starts and never returned. Flink assigns this managed memory to its algorithms and the algorithms spill to disk if the amount of data exceeds their memory budget. Since all memory is pre-allocated and internally managed by Flink, it is not possible to measure the actual memory consumption.