Flink常用配置参数

taskmanager.memory.size

TaskManager在JVM堆内存中保留空间的大小，此内存用于排序，哈希表和中间状态的缓存。如果未指定，则会使用JVM堆内存乘以比例taskmanager.memory.fraction。单位：MB。

0

否

仅MRS 3.x之前版本

taskmanager.registration.initial-backoff

两次连续注册的初始间隔时间。单位：ms/s/m/h/d。

时间数值和单位之间有半角字符空格。ms/s/m/h/d表示毫秒、秒、分钟、小时、天。

500 ms

否

taskmanager.registration.refused-backoff

JobManager拒绝注册后到允许再次注册的间隔时间。

5 min

否

taskmanager.rpc.port

TaskManager的IPC端口范围。

32326-32390

否

适用于所有版本

taskmanager.memory.segment-size

内存管理器和网络堆栈使用的内存缓冲区大小。单位：bytes。

32768

否

taskmanager.data.port

TaskManager数据交换端口范围。

32391-32455

否

taskmanager.data.ssl.enabled

TaskManager之间数据传输是否使用SSL加密，仅在全局开关security.ssl开启时有效。

false

否

taskmanager.numberOfTaskSlots

TaskManager占用的slot数，一般配置成物理机的核数，yarn-session模式下只能使用-s参数传递，yarn-cluster模式下只能使用-ys参数传递。

1

否

parallelism.default

默认并行度，用于未指定并行度的作业。

1

否

taskmanager.memory.fraction

TaskManager在JVM堆内存中保留空间的比例，此内存用于排序，哈希表和中间状态的缓存。

0.7

否

taskmanager.memory.off-heap

TaskManager是否使用堆外内存，此内存用于排序，哈希表和中间状态的缓存。建议对于大内存，开启此配置提高内存操作的效率。

false

是

taskmanager.memory.preallocate

TaskManager是否在启动时分配保留内存空间。当开启堆外内存时，建议开启此配置项。

false

否

task.cancellation.interval

两次连续任务取消操作的间隔时间。单位：ms。

30000

否

client.rpc.port

Flink client端Akka system监测端口。

32651-32720

否

仅MRS 3.x及之后版本

jobmanager.heap.size

JobManager堆内存大小，yarn-session模式下只能使用-jm参数传递，yarn-cluster模式下只能使用-yjm参数传递，如果小于YARN配置文件中yarn.scheduler.minimum-allocation-mb大小，则使用YARN配置中的值。单位：B/KB/MB/GB/TB。

1024mb

否

taskmanager.heap.size

TaskManager堆内存大小，yarn-session模式下只能使用-tm参数传递，yarn-cluster模式下只能使用-ytm参数传递，如果小于YARN配置文件中yarn.scheduler.minimum-allocation-mb大小，则使用YARN配置中的值。单位：B/KB/MB/GB/TB。

1024mb

否

taskmanager.network.numberOfBuffers

TaskManager网络传输缓冲栈数量，如果作业运行中出错提示系统中可用缓冲不足，可以增加这个配置项的值。

2048

否

taskmanager.debug.memory.startLogThread

调试Flink内存和GC相关问题时可开启，TaskManager会定时采集内存和GC的统计信息，包括当前堆内，堆外，内存池的使用率和GC时间。

false

否

taskmanager.debug.memory.logIntervalMs

TaskManager定时采集内存和GC的统计信息的采集间隔。

0

否

taskmanager.maxRegistrationDuration

TaskManager向JobManager注册自己的最长时间，如果超过时间，TaskManager会关闭。

5 min

否

taskmanager.initial-registration-pause

两次连续注册的初始间隔时间。该值需带一个时间单位（ms/s/min/h/d）（比如5秒）。

时间数值和单位之间有半角字符空格。ms/s/m/h/d表示毫秒、秒、分钟、小时、天。

500 ms

否

taskmanager.max-registration-pause

TaskManager注册失败最大重试间隔。单位：ms/s/m/h/d。

30 s

否

taskmanager.refused-registration-pause

TaskManager注册连接被JobManager拒绝后的重试间隔。单位：ms/s/m/h/d。

10 s

否

classloader.resolve-order

从用户代码加载类时定义类解析策略，这意味着是首先检查用户代码jar（“child-first”）还是应用程序类路径（“parent-first”）。默认设置指示首先从用户代码jar加载类，这意味着用户代码jar可以包含和加载不同于Flink使用的（依赖）依赖项。

child-first

否

slot.idle.timeout

Slot Pool中空闲Slot的超时时间（以ms为单位）。

50000

否

slot.request.timeout

从Slot Pool请求Slot的超时（以ms为单位）。

300000

否

task.cancellation.timeout

取消任务超时时间（以ms为单位），超时后会触发TaskManager致命错误。设置为0，取消任务卡住则不会报错。

180000

否

taskmanager.network.detailed-metrics

启用网络队列长度的详细指标监控。

false

否

taskmanager.network.memory.buffers-per-channel

每个传出/传入通道（子分区/输入通道）使用的最大网络缓冲区数.在基于信用的流量控制模式下，这表示每个输入通道中有多少信用。它应配置至少2以获得良好的性能。1个缓冲区用于接收子分区中的飞行中数据，1个缓冲区用于并行序列化。

2

否

taskmanager.network.memory.floating-buffers-per-gate

每个输出/输入门（结果分区/输入门）使用的额外网络缓冲区数。在基于信用的流量控制模式中，这表示在所有输入通道之间共享多少浮动信用。浮动缓冲区基于积压（子分区中的实时输出缓冲区）反馈来分布，并且可以帮助减轻由子分区之间的不平衡数据分布引起的背压。如果节点之间的往返时间较长和/或群集中的机器数量较多，则应增加此值。

8

否

taskmanager.network.memory.fraction

用于网络缓冲区的JVM内存的占比。这决定了TaskManager可以同时拥有多少流数据交换通道以及通道缓冲的程度。如果作业被拒绝或者收到系统没有足够缓冲区的警告，请增加此值或“taskmanager.network.memory.min”和“taskmanager.network.memory.max”。另请注意，“taskmanager.network.memory.min”和“taskmanager.network.memory.max”可能会覆盖此占比。

0.1

否

taskmanager.network.memory.max

网络缓冲区的最大内存大小。该值需带一个大小单位（B/KB/MB/GB/TB）。

1 GB

否

taskmanager.network.memory.min

网络缓冲区的最小内存大小。该值需带一个大小单位（B/KB/MB/GB/TB）。

64 MB

否

taskmanager.network.request-backoff.initial

输入通道的分区请求的最小退避（以ms为单位）。

100

否

taskmanager.network.request-backoff.max

输入通道的分区请求的最大退避（以ms为单位）。

10000

否

taskmanager.registration.timeout

TaskManager注册的超时时间，在该时间内未成功注册，TaskManager将终止。该值需带一个时间单位（ms/s/min/h/d）。

5 min

否

resourcemanager.taskmanager-timeout

释放空闲TaskManager的超时（以ms为单位）。

30000

否

blob.server.port

blob服务器端口。

32456-32520

否

blob.service.ssl.enabled

blob传输通道是否加密传输，仅在全局开关security.ssl开启时有。

true

是

blob.fetch.retries

TaskManager从JobManager下载blob文件的重试次数。

50

否

blob.fetch.num-concurrent

JobManager支持的下载blob的并发数。

50

否

blob.fetch.backlog

JobManager支持的blob下载队列大小，比如下载Jar包等。单位：个。

1000

否

library-cache-manager.cleanup.interval

当用户取消flink job后，jobmanager删除HDFS上存放用户jar包的时间，单位为s。

仅适用于MRS 3.x及之后版本。

3600

否

akka.ask.timeout

akka所有异步请求和阻塞请求的超时时间。如果Flink发生超时失败，可以增大这个值。当机器处理速度慢或者网络阻塞时会发生超时。单位：ms/s/m/h/d。

10s

否

适用于所有版本

akka.lookup.timeout

查找JobManager actor对象的超时时间。单位：ms/s/m/h/d。

10s

否

akka.framesize

JobManager和TaskManager间最大消息传输大小。当Flink出现消息大小超过限制的错误时，可以增大这个值。单位：b/B/KB/MB。

10485760b

否

akka.watch.heartbeat.interval

Akka DeathWatch机制检测失联TaskManager的心跳间隔。如果TaskManager经常发生由于心跳消息丢失或延误而被错误标记为失联的情况，可以增大这个值。单位：ms/s/m/h/d。

10s

否

akka.watch.heartbeat.pause

Akka DeathWatch可接受的心跳暂停时间，较小的数值表示不允许不规律的心跳。单位：ms/s/m/h/d。

60s

否

akka.watch.threshold

DeathWath失败检测阈值，较小的数值容易把正常TaskManager标记为失败，较大的值增加了失败检测的时间。

12

否

akka.tcp.timeout

发送连接TCP超时时间，如果经常发生满网络环境下连接TaskManager超时，可以增大这个值。单位：ms/s/m/h/d。

20s

否

akka.throughput

Akka批量处理消息的数量，一次操作完后把处理线程归还线程池。较小的数值代表actor消息处理的公平调度，较大的值以牺牲调度公平的代价提高整体性能。

15

否

akka.log.lifecycle.events

Akka远程时间日志开关，当需要调试时可打开此开关。

false

否

akka.startup-timeout

远程组件启动失败前的超时时间。该值需带一个时间单位（ms/s/min/h/d）

与akka.ask.timeout的值一致

否

akka.ssl.enabled

Akka通信SSL开关，仅在全局开关security.ssl开启时有。

true

是

akka.client-socket-worker-pool.pool-size-factor

计算线程池大小的因子，计算公式：ceil（可用处理器*因子），计算结果限制在pool-size-min和pool-size-max之间。

1.0

否

仅适用于MRS 3.x及之后版本

akka.client-socket-worker-pool.pool-size-max

基于因子计算的线程数上限。

2

否

akka.client-socket-worker-pool.pool-size-min

基于因子计算的线程数下限。

1

否

akka.client.timeout

【说明】客户端超时时间。该值需带一个时间单位（ms/s/min/h/d）。

60s

否

akka.server-socket-worker-pool.pool-size-factor

【说明】计算线程池大小的因子，计算公式：ceil（可用处理器*因子），计算结果限制在pool-size-min和pool-size-max之间。

1.0

否

akka.server-socket-worker-pool.pool-size-max

基于因子计算的线程数上限。

2

否

akka.server-socket-worker-pool.pool-size-min

基于因子计算的线程数下限。

1

否

security.ssl.internal.enabled

内部通信SSL总开关，按照集群的安全模式自动配置。

• 安全模式：true
• 普通模式：false

是

仅MRS 3.x之前版本

security.ssl.internal.keystore

Java keystore文件。

-

是

security.ssl.internal.keystore-password

keystore文件解密密码。

-

是

security.ssl.internal.key-password

keystore文件中服务端key的解密密码。

-

是

security.ssl.internal.truststore

truststore文件包含公共CA证书。

-

是

security.ssl.internal.truststore-password

truststore文件解密密码。

-

是

security.ssl.rest.enabled

外部通信SSL总开关，按照集群的安全模式自动配置。

• 安全模式：true
• 普通模式：false

是

security.ssl.rest.keystore

Java keystore文件。

-

是

security.ssl.rest.keystore-password

keystore文件解密密码。

-

是

security.ssl.rest.key-password

keystore文件中服务端key的解密密码。

-

是

security.ssl.rest.truststore

truststore文件包含公共CA证书。

-

是

security.ssl.rest.truststore-password

truststore文件解密密码。

-

是

security.ssl.protocol

SSL传输的协议版本。

TLSv1.2

是

适用于所有版本

security.ssl.algorithms

支持的SSL标准算法，具体可参考java官网：http://docs.oracle.com/javase/8/docs/technotes/guides/security/StandardNames.html#ciphersuites。

TLS_DHE_RSA_WITH_AES_128_GCM_SHA256,TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,TLS_DHE_RSA_WITH_AES_256_GCM_SHA384,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384

是

security.ssl.enabled

内部通信SSL总开关，按照集群的安装模式自动配置。

• 安全模式：true
• 普通模式：false

是

仅MRS 3.x及之后版本

security.ssl.keystore

Java keystore文件。

-

是

security.ssl.keystore-password

keystore文件解密密码。

-

是

security.ssl.key-password

keystore文件中服务端key的解密密码。

-

是

security.ssl.truststore

truststore文件包含公共CA证书。

-

是

security.ssl.truststore-password

truststore文件解密密码。

-

是

taskmanager.network.netty.num-arenas

Netty内存块数。

1

否

taskmanager.network.netty.server.numThreads

Netty服务器线程的数量。

1

否

taskmanager.network.netty.client.numThreads

Netty客户端线程数。

1

否

taskmanager.network.netty.client.connectTimeoutSec

Netty客户端连接超时。单位：s。

120

否

taskmanager.network.netty.sendReceiveBufferSize

Netty发送和接收缓冲区大小。 默认为系统缓冲区大小（cat / proc / sys / net / ipv4 / tcp_ [rw] mem），在现代Linux中为4MB。单位：bytes。

4096

否

taskmanager.network.netty.transport

Netty传输类型，“nio”或“epoll”。

nio

否

jobmanager.web.port

web端口，支持范围：32261-32325。

32261-32325

否

仅MRS 3.x之前版本

jobmanager.web.allow-access-address

web访问白名单，ip以逗号隔开。只有在白名单中的ip才能访问web。

*

是

适用于所有版本

flink.security.enable

用户安装Flink集群时，需要选择“安全模式”或“普通模式”。

• 当选择“安全模式”，自动配置为“true”。
• 当选择“普通模式”，自动配置为“false”。

对于已经安装好的Flink集群，用户可以通过查看配置的值来区分当前安装的是安全模式还是普通模式。

自动配置

否

仅MRS 3.x及之后版本

rest.bind-port

web端口，支持范围：32261-32325。

32261-32325

否

jobmanager.web.history

显示“flink.security.enable”最近的job数目。

5

否

jobmanager.web.checkpoints.disable

禁用checkpoint统计。

false

否

jobmanager.web.checkpoints.history

Checkpoint统计记录数。

10

否

jobmanager.web.backpressure.cleanup-interval

未访问反压记录清理周期。单位：ms。

600000

否

jobmanager.web.backpressure.refresh-interval

反压记录刷新周期。单位：ms。

60000

否

jobmanager.web.backpressure.num-samples

计算反压使用的堆栈跟踪记录数。

100

否

jobmanager.web.backpressure.delay-between-samples

计算反压的采样间隔。单位：ms

50

否

jobmanager.web.ssl.enabled

web是否使用SSL加密传输，仅在全局开关security.ssl开启时有。

false

是

jobmanager.web.accesslog.enable

web操作日志使能开关，日志会存放在webaccess.log中。

true

是

jobmanager.web.x-frame-options

http安全头X-Frame-Options的值，可选范围为：SAMEORIGIN、DENY、ALLOW-FROM uri。

DENY

是

jobmanager.web.cache-directive

web页面是否支持缓存。

no-store：所有内容都不会被保存到缓存

是

jobmanager.web.expires-time

web页面缓存过期时长。单位：ms。

0

是

jobmanager.web.access-control-allow-origin

网页同源策略，防止跨域攻击。*表示允许任意网站跨域访问该服务端口，可配置为指定网址。

*（非安全集群）

是

jobmanager.web.refresh-interval

web网页刷新时间。单位：ms。

3000

是

jobmanager.web.logout-timer

配置无操作情况下自动登出时间间隔。单位：ms。

600000

是

jobmanager.web.403-redirect-url

web403页面，访问若遇到403错误，则会重定向到配置的页面。

自动配置

是

jobmanager.web.404-redirect-url

web404页面，访问若遇到404错误，则会重定向到配置的页面。

自动配置

是

jobmanager.web.415-redirect-url

web415页面，访问若遇到415错误，则会重定向到配置的页面。

自动配置

是

jobmanager.web.500-redirect-url

web500页面，访问若遇到500错误，则会重定向到配置的页面。

自动配置

是

rest.await-leader-timeout

客户端等待Leader地址的时间（以ms为单位）。

30000

否

rest.client.max-content-length

客户端处理的最大内容长度（以字节为单位）。

104857600

否

rest.connection-timeout

客户端建立TCP连接的最长时间（以ms为单位）。

15000

否

rest.idleness-timeout

连接保持空闲状态的最长时间（以ms为单位）。

300000

否

rest.retry.delay

客户端在连续重试之间等待的时间（以ms为单位）。

3000

否

rest.retry.max-attempts

如果可重试算子操作失败，客户端将尝试重试的次数。

20

否

rest.server.max-content-length

服务端处理的最大内容长度（以字节为单位）。

104857600

否

rest.server.numThreads

异步处理请求的最大线程数。

4

否

web.timeout

web监控超时时间（以ms为单位）。

10000

否

fs.overwrite-files

文件输出写操作是否默认覆盖已有文件。

false

否

fs.output.always-create-directory

当文件写入程序的并行度大于1时，输出文件的路径下会创建一个目录，并将不同的结果文件（每个并行写程序任务）放入该目录。

• 设置为true，那么并行度为1的写入程序也将创建一个目录并将一个结果文件放入其中。
• 设置为false，则并行度为1的写入程序将直接在输出路径中创建文件，而不再创建目录。

false

否

state.backend.fs.checkpointdir

当backend为filesystem时的路径，路径必须能够被JobManager访问到，本地路径只支持local模式，集群模式下请使用HDFS路径。

hdfs:///flink/checkpoints

否

state.savepoints.dir

Flink用于恢复和更新作业的保存点存储目录。当触发保存点的时候，保存点元数据信息将会保存到该目录中。

hdfs:///flink/savepoint

安全模式下必配

restart-strategy

默认重启策略，用于未指定重启策略的作业：

• fixed-delay
• failure-rate
• none

none

否

restart-strategy.fixed-delay.attempts

fixed-delay策略重试次数。

• 作业中开启了checkpoint，默认值为Integer.MAX_VALUE。
• 作业中未开启checkpoint，默认值为3。

否

restart-strategy.fixed-delay.delay

fixed-delay策略重试间隔时间。单位：ms/s/m/h/d。

• 作业中开启了checkpoint，默认值是10 s。
• 作业中未开启checkpoint，默认值和配置项akka.ask.timeout的值一致。

否

restart-strategy.failure-rate.max-failures-per-interval

故障率策略下作业失败前给定时间段内的最大重启次数。

1

否

restart-strategy.failure-rate.failure-rate-interval

failure-rate策略重试时间。单位：ms/s/m/h/d。

60 s

否

restart-strategy.failure-rate.delay

failure-rate策略重试间隔时间。单位：ms/s/m/h/d。

默认值和akka.ask.timeout配置值一样。可参考Distributed Coordination (via Akka)。

否

security.kerberos.login.keytab

该参数为客户端参数，keytab路径。

根据实际业务配置

是

security.kerberos.login.principal

该参数为客户端参数，如果keytab和principal都设置，默认会使用keytab认证。

根据实际业务配置

否

security.kerberos.login.contexts

该参数为服务器端参数，flink生成jass文件的contexts。

Client、KafkaClient

是

high-availability

HA模式，是启用HA还是非HA模式。当前支持两种模式：

• none，只运行单个jobManager，jobManager的状态不进行Checkpoint。
• ZooKeeper。
  • 非YARN模式下，支持多个jobManager，通过选举产生leader。
  • YARN模式下只存在一个jobManager。

zookeeper

否

high-availability.zookeeper.quorum

ZooKeeper quorum地址。

自动配置

否

high-availability.zookeeper.path.root

Flink在ZooKeeper上创建的根目录，存放HA模式必须的元数据。

/flink

否

high-availability.storageDir

存放state backend中JobManager元数据，ZooKeeper只保存实际数据的指针。

hdfs:///flink/recovery

否

high-availability.zookeeper.client.session-timeout

ZooKeeper客户端会话超时时间。单位：ms。

60000

否

high-availability.zookeeper.client.connection-timeout

ZooKeeper客户端连接超时时间。单位：ms。

15000

否

high-availability.zookeeper.client.retry-wait

ZooKeeper客户端重试等待时间。单位：ms。

5000

否

high-availability.zookeeper.client.max-retry-attempts

ZooKeeper客户端最大重试次数。

3

否

high-availability.job.delay

当jobManager恢复后重启job的延迟时间。

仅适用于MRS 3.x及之后版本。

默认值和akka.ask.timeout配置值保持一致

否

high-availability.zookeeper.client.acl

设置ZooKeeper节点的ACL (open creator)，按照集群的安全模式自动配置。设置ACL选项请参考：https://zookeeper.apache.org/doc/r3.5.1-alpha/zookeeperProgrammers.html#sc_BuiltinACLSchemes。

• 安全模式：creator
• 非安全模式：open

是

zookeeper.sasl.disable

基于SASL认证的使能开关，按照集群的安全模式自动配置：。

• 安全模式：false
• 非安全模式：true

是

zookeeper.sasl.service-name

• 如果ZooKeeper服务端配置了不同于“ZooKeeper”的服务名，可以设置此配置项。
• 如果客户端和服务端的服务名不一致，认证会失败。

zookeeper

是

env.java.opts

JVM参数，会传递到启动脚本，JobManager，TaskManager，Yarn客户端。比如传递远程调试的参数等。

-Xloggc:<LOG_DIR>/gc.log -XX:+PrintGCDetails -XX:-OmitStackTraceInFastThrow -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=20 -XX:GCLogFileSize=20M -Djdk.tls.ephemeralDHKeySize=2048 -Djava.library.path=${HADOOP_COMMON_HOME}/lib/native -Djava.net.preferIPv4Stack=true -Djava.net.preferIPv6Addresses=false -Dbeetle.application.home.path=/opt/xxx/Bigdata/common/runtime/security/config

否

yarn.maximum-failed-containers

当TaskManager所属容器出错后，重新申请container次数。默认值为Flink集群启动时TaskManager的数量。

5

否

yarn.application-attempts

Application master重启次数，次数是算在一个validity interval的最大次数，validity interval在flink中设置为akka的timeout。重启后AM的地址和端口会变化，client需要手动连接。

2

否

yarn.heartbeat-delay

Application Master和YARN Resource Manager心跳的时间间隔。单位：seconds

5

否

yarn.containers.vcores

每个Yarn容器的虚拟核数。

TaskManager的slot数

否

yarn.application-master.port

Application Master端口号设置，支持端口范围。

32586-32650

否

nettyconnector.registerserver.topic.storage

设置NettySink的IP、端口及并发度信息在第三方注册服务器上的路径。建议用户使用ZooKeeper进行存储。

/flink/nettyconnector

否，当使用pipeline特性为必选

nettyconnector.sinkserver.port.range

设置NettySink的端口范围。

28444-28843

否，当使用pipeline特性为必选

nettyconnector.ssl.enabled

设置NettySink与NettySource之间通信是否配置SSL加密。其中加密密钥以及加密协议等请参见SSL。

false

否，当使用pipeline特性为必选

nettyconnector.message.delimiter

用来配置nettysink发送给nettysource消息的分隔符，长度为2-4个字节，不可包含“\n”， “ ”， “#” 。

默认使用“$_”

否，当使用pipeline特性为必选