Flink内存优化_优化flink内存占用

flink 分别提供了通用和细粒度的内存配置，来满足不同用户的需求。

一、Flink内存模型

• Total Process Memory: 包括 flink 应用消耗的内存(Total Flink Memory) 和 JVM 消耗的内存
• Total Flink Memory: 包括 JVM heap, managed memory 和 direct memory

如果用户只是使用了 local 模式运行 flink(比如通过idea)，那么只有部分内存配置是生效的，具体配置后面有讲到。

对于集群上运行的大部分程序来说，最简单的是配置下面任意一个：

• Total Flink memory (taskmanager.memory.flink.size)
• Total process memory (taskmanager.memory.process.size)

其余的内存组件将根据默认值或额外配置的选项自动调整。关于其他内存组件的更多细节，可以阅读下一章节。

对于 standalone 部署模式来说，因为我们要声明分配给 flink 自身的内存大小，所以配置 Total Flink memory 更合适。

对于容器化部署模式来说，配置 Total Process Memory 更好。因为 Total Process Memory 实际上代表 Flink JVM 进程占用的总内存，也是容器允许 TM 占用内存的上线，超过这个值就会被容器杀掉。

另一种方法是配置 task heap (taskmanager.memory.task.heap.size) 和 managed memory(taskmanager.memory.managed.size). 这是一种细粒度的配置方式。

注意：上面提到的三种配置方式，必须至少选择一种进行配置(官方建议只选一种)，否则 Flink 会启动失败。这三种配置没有默认值，必须选择一种配置：

• taskmanager.memory.flink.size
• taskmanager.memory.process.size
• taskmanager.memory.task.heap.size and taskmanager.memory.managed.size

注意：不建议同时显式地配置 Total Flink memory 和整个 Total Process Memory 。由于潜在的内存配置冲突，它可能导致部署失败。其他内存组件的额外配置也需要谨慎，因为它可能会产生更多的配置冲突。

二、配置 Task Heap 和 Managed Memory

除了上一小节提到的总内存配置，用户可以显示的指定 task heap 和 managed memory。例如当我们需要确定的内存时，可以直接通过这种配置指定，不再需要像之前那样，从总内存中计算出 task heap.
其余的内存组件将根据默认值或额外配置的选项自动调整。

2.1、Task (Operator) Heap Memory

如果你想确保你的代码有一定数量的JVM堆可用，可以显式地设置任务堆内存(taskmanager.memory.task.heap.size)。它将被添加到JVM堆大小中，并将专用于运行用户代码的Flink操作符。
直接影响 task executor 的 -Xmx and -Xms 参数:

-Xmx and -Xms  =  flink framework Heap(默认 128M) + Task Heap

2.2、Managed Memory

Managed memory 由 flink 管理，并且使用的 native memory (off-heap). 使用 Managed memory 的有：

• Streaming 作业的 RocksDB state backend
• Batch 作业可以使用它对中间结果进行排序(sort)、散列表(hash)和缓存(caching )

Managed memory 有两种配置方法：

• 直接 taskmanager.memory.managed.size 配置
• 计算 Total Flink Memory * taskmanager.memory.managed.fraction(默认0.4)

taskmanager.memory.managed.size 会覆盖通过 fraction 计算得出的大小

三、配置 Off-Heap Memory (direct or native)

用户申请的 off-heap 被算做 task off-heap memory，通过 taskmanager.memory.task.off-heap.size 配置。
注意：用户也可以调整 framework off-heap memory，即 flink 框架使用的堆外内存。这个是高级配置，最好确定需要时才进行调整。

flink 将 framework off-heap memory 和 task off-heap memory 纳入 JVM 的 direct memory 限制参数中：

-XX:MaxDirectMemorySize = Framework  Off-heap + Task Off-Heap + Network Memory

四、配置flink-conf.yaml示例 

# JobManager runs.
 
jobmanager.rpc.address: www.slave2.asiainfo.com
 
# The RPC port where the JobManager is reachable.
 
jobmanager.rpc.port: 6123
 
 
# The heap size for the JobManager JVM
 
#jobmanager.heap.size: 2048m
 
 
# The total process memory size for the TaskManager.
#
# Note this accounts for all memory usage within the TaskManager process, including JVM metaspace and other overhead.
 
#taskmanager.memory.process.size: 4096m
 
# To exclude JVM metaspace and overhead, please, use total Flink memory size instead of 'taskmanager.memory.process.size'.
# It is not recommended to set both 'taskmanager.memory.process.size' and Flink memory.
#
taskmanager.memory.flink.size: 16384m
 
taskmanager.memory.jvm-metaspace.size: 1024m
 
# The number of task slots that each TaskManager offers. Each slot runs one parallel pipeline.
 
taskmanager.numberOfTaskSlots: 4
 
# The parallelism used for programs that did not specify and other parallelism.
 
parallelism.default: 4
 
# The default file system scheme and authority.
#
# By default file paths without scheme are interpreted relative to the local
# root file system 'file:///'. Use this to override the default and interpret
# relative paths relative to a different file system,
# for example 'hdfs://mynamenode:12345'
#
# fs.default-scheme
 
#==============================================================================
# High Availability
#==============================================================================
 
# The high-availability mode. Possible options are 'NONE' or 'zookeeper'.
#
# high-availability: zookeeper
 
# The path where metadata for master recovery is persisted. While ZooKeeper stores
# the small ground truth for checkpoint and leader election, this location stores
# the larger objects, like persisted dataflow graphs.
#
# Must be a durable file system that is accessible from all nodes
# (like HDFS, S3, Ceph, nfs, ...)
#
# high-availability.storageDir: hdfs:///flink/ha/
 
# The list of ZooKeeper quorum peers that coordinate the high-availability
# setup. This must be a list of the form:
# "host1:clientPort,host2:clientPort,..." (default clientPort: 2181)
#
# high-availability.zookeeper.quorum: localhost:2181
 
 
# ACL options are based on https://zookeeper.apache.org/doc/r3.1.2/zookeeperProgrammers.html#sc_BuiltinACLSchemes
# It can be either "creator" (ZOO_CREATE_ALL_ACL) or "open" (ZOO_OPEN_ACL_UNSAFE)
# The default value is "open" and it can be changed to "creator" if ZK security is enabled
#
# high-availability.zookeeper.client.acl: open
 
#==============================================================================
# Fault tolerance and checkpointing
#==============================================================================
 
# The backend that will be used to store operator state checkpoints if
# checkpointing is enabled.
#
# Supported backends are 'jobmanager', 'filesystem', 'rocksdb', or the
# <class-name-of-factory>.
#
# state.backend: filesystem
 
# Directory for checkpoints filesystem, when using any of the default bundled
# state backends.
#
# state.checkpoints.dir: hdfs://namenode-host:port/flink-checkpoints
 
# Default target directory for savepoints, optional.
#
# state.savepoints.dir: hdfs://namenode-host:port/flink-checkpoints
 
# Flag to enable/disable incremental checkpoints for backends that
# support incremental checkpoints (like the RocksDB state backend).
#
# state.backend.incremental: false
 
# The failover strategy, i.e., how the job computation recovers from task failures.
# Only restart tasks that may have been affected by the task failure, which typically includes
# downstream tasks and potentially upstream tasks if their produced data is no longer available for consumption.
 
jobmanager.execution.failover-strategy: region
 
#==============================================================================
# Rest & web frontend
#==============================================================================
 
# The port to which the REST client connects to. If rest.bind-port has
# not been specified, then the server will bind to this port as well.
#
#rest.port: 8081
 
# The address to which the REST client will connect to
#
#rest.address: 0.0.0.0
 
# Port range for the REST and web server to bind to.
#
#rest.bind-port: 8080-8090
 
# The address that the REST & web server binds to
#
#rest.bind-address: 0.0.0.0
 
# Flag to specify whether job submission is enabled from the web-based
# runtime monitor. Uncomment to disable.
 
#web.submit.enable: false
 
#==============================================================================
# Advanced
#==============================================================================
 
# Override the directories for temporary files. If not specified, the
# system-specific Java temporary directory (java.io.tmpdir property) is taken.
#
# For framework setups on Yarn or Mesos, Flink will automatically pick up the
# containers' temp directories without any need for configuration.
#
# Add a delimited list for multiple directories, using the system directory
# delimiter (colon ':' on unix) or a comma, e.g.:
#     /data1/tmp:/data2/tmp:/data3/tmp
#
# Note: Each directory entry is read from and written to by a different I/O
# thread. You can include the same directory multiple times in order to create
# multiple I/O threads against that directory. This is for example relevant for
# high-throughput RAIDs.
#
# io.tmp.dirs: /tmp
 
# The classloading resolve order. Possible values are 'child-first' (Flink's default)
# and 'parent-first' (Java's default).
#
# Child first classloading allows users to use different dependency/library
# versions in their application than those in the classpath. Switching back
# to 'parent-first' may help with debugging dependency issues.
#
# classloader.resolve-order: child-first
 
# The amount of memory going to the network stack. These numbers usually need
# no tuning. Adjusting them may be necessary in case of an "Insufficient number
# of network buffers" error. The default min is 64MB, the default max is 1GB.
#
taskmanager.memory.network.fraction: 0.5
# taskmanager.memory.network.min: 256mb
# taskmanager.memory.network.max: 4gb
 
#==============================================================================
# Flink Cluster Security Configuration
#==============================================================================
 
# Kerberos authentication for various components - Hadoop, ZooKeeper, and connectors -
# may be enabled in four steps:
# 1. configure the local krb5.conf file
# 2. provide Kerberos credentials (either a keytab or a ticket cache w/ kinit)
# 3. make the credentials available to various JAAS login contexts
# 4. configure the connector to use JAAS/SASL
 
# The below configure how Kerberos credentials are provided. A keytab will be used instead of
# a ticket cache if the keytab path and principal are set.
 
# security.kerberos.login.use-ticket-cache: true
# security.kerberos.login.keytab: /path/to/kerberos/keytab
# security.kerberos.login.principal: flink-user
 
# The configuration below defines which JAAS login contexts
 
# security.kerberos.login.contexts: Client,KafkaClient
 
#==============================================================================
# ZK Security Configuration
#==============================================================================
 
# Below configurations are applicable if ZK ensemble is configured for security
 
# Override below configuration to provide custom ZK service name if configured
# zookeeper.sasl.service-name: zookeeper
 
# The configuration below must match one of the values set in "security.kerberos.login.contexts"
# zookeeper.sasl.login-context-name: Client
 
#==============================================================================
# HistoryServer
#==============================================================================
 
# The HistoryServer is started and stopped via bin/historyserver.sh (start|stop)
 
# Directory to upload completed jobs to. Add this directory to the list of
# monitored directories of the HistoryServer as well (see below).
#jobmanager.archive.fs.dir: hdfs:///completed-jobs/
 
# The address under which the web-based HistoryServer listens.
#historyserver.web.address: 0.0.0.0
 
# The port under which the web-based HistoryServer listens.
#historyserver.web.port: 8082
 
# Comma separated list of directories to monitor for completed jobs.
#historyserver.archive.fs.dir: hdfs:///completed-jobs/
 
# Interval in milliseconds for refreshing the monitored directories.
#historyserver.archive.fs.refresh-interval: 10000