FLink配置
配置
所有配置都在conf/flink-conf.yaml中完成，这个文件应该是一个扁平的YAML键值对集合，格式为key: value。配置在Flink进程启动时被解析和评估。对配置文件的更改需要重新启动相关的进程。
默认配置将使用您的默认Java安装。如果要手动覆盖Java运行时，请在conf/flink-conf.yaml中手动设置环境变量JAVA_HOME或配置键env.java.home。
您可以通过定义FLINK_CONF_DIR环境变量来指定不同的配置目录位置。对于提供非会话部署的资源提供者，您可以以这种方式指定每个作业的配置。从Flink发行版中复制conf目录并根据每个作业的需求修改设置。请注意，这在Docker或独立Kubernetes部署中不受支持。在基于Docker的部署中，您可以使用FLINK_PROPERTIES环境变量传递配置值。
在会话集群上，提供的配置仅用于配置执行参数，例如影响作业的配置参数，而不是底层集群。
基本设置
默认配置支持启动一个单节点的Flink会话集群，无需进行任何更改。本节中的选项是基本分布式Flink设置中最常用的选项。
主机名/端口
这些选项仅适用于独立应用程序或会话部署（简单独立或Kubernetes）。
如果您使用带有Yarn或active Kubernetes集成的Flink，则主机名和端口将自动发现。
-
rest.address，rest.port：这些由客户端用于连接到Flink。将其设置为JobManager运行的主机名或JobManager REST接口前端的（Kubernetes）服务的主机名。 -
jobmanager.rpc.address（默认为“localhost”）和jobmanager.rpc.port（默认为“6123”）配置条目用于TaskManager连接到JobManager/ResourceManager。将其设置为JobManager运行的主机名或JobManager的（Kubernetes内部）服务的主机名。在使用高可用性的设置中，此选项将被忽略，因为会使用领导者选举机制自动发现。
内存大小
默认的内存大小支持简单的流处理/批处理应用程序，但对于更复杂的应用程序来说，性能太低。
-
jobmanager.memory.process.size：JobManager（JobMaster / ResourceManager / Dispatcher）进程的总大小。 -
taskmanager.memory.process.size：TaskManager进程的总大小。
总大小包括所有内容。Flink将减去一些内存以满足JVM自身的内存需求（元空间和其他），并在其组件之间自动划分和配置剩余的内存（JVM堆、非堆、对于任务管理器还有网络、托管内存等）。
这些值是配置为内存大小的，例如1536m或2g。
并行度
-
taskmanager.numberOfTaskSlots：TaskManager提供的插槽数（默认值：1）。每个插槽可以执行一个任务或流水线。在一个TaskManager中使用多个插槽可以帮助分摊某些常量开销（JVM、应用程序库或网络连接）到并行任务或流水线上。有关详细信息，请参见任务插槽和资源概念部分。运行多个具有一个插槽的较小的TaskManager是一个好的起点，并且可以实现最佳的任务隔离。将相同的资源分配给较少的具有更多插槽的较大TaskManager可以增加资源利用率，但牺牲了任务之间的隔离性（更多的任务共享相同的JVM）。
-
parallelism.default：当没有在任何地方指定并行度时使用的默认并行度（默认值：1）。
检查点
您可以在Flink作业或应用程序的代码中直接配置检查点。将这些值放在配置文件中定义后，如果应用程序未进行任何配置，则会将其作为默认值。
-
state.backend.type：要使用的状态后端。这定义了用于创建快照的数据结构机制。常见的值为filesystem或rocksdb。 -
state.checkpoints.dir：写入检查点的目录。它采用路径URI，例如s3://mybucket/flink-app/checkpoints或hdfs://namenode:port/flink/checkpoints。 -
state.savepoints.dir：保存点的默认目录。采用路径URI，类似于state.checkpoints.dir。 -
execution.checkpointing.interval：基本间隔设置。要启用检查点，您需要将此值设置为大于0。
Web界面
-
web.submit.enable：启用通过Flink UI上传和启动作业（默认为true）。请注意，即使禁用此选项，会话集群仍然通过REST请求（HTTP调用）接受作业。此标志仅保护在UI中上传作业的功能。 -
web.cancel.enable：启用通过Flink UI取消作业（默认为true）。请注意，即使禁用此选项，会话集群仍然通过REST请求（HTTP调用）取消作业。此标志仅保护在UI中取消作业的功能。 -
web.upload.dir：存储上传作业的目录。仅当web.submit.enable为true时使用。 -
web.exception-history-size：设置打印由Flink处理的作业的最近故障的异常历史记录的大小。
其他
-
io.tmp.dirs：Flink放置本地数据的目录，默认为系统临时目录（java.io.tmpdir属性）。如果配置了目录列表，则Flink将文件在这些目录之间进行轮转。这些目录中的数据默认包括RocksDB创建的文件、溢出的中间结果（批处理算法）和缓存的jar文件。
这些数据不依赖于持久性/恢复，但如果删除了这些数据，通常会导致一次重量级的恢复操作。因此，建议将其设置为不会自动定期清除的目录。
Yarn和Kubernetes设置默认将此值配置为本地工作目录。
常见设置选项
用于配置Flink应用程序或集群的常见选项。
主机和端口
用于配置不同Flink组件的主机名和端口的选项。
仅在没有高可用性的独立设置中，JobManager的主机名和端口才是相关的。在这种设置中，任务管理器使用这些配置值来查找（和连接到）JobManager。在所有高可用性的设置中，TaskManagers通过高可用性服务（例如ZooKeeper）发现JobManager。
使用资源编排框架（K8s、Yarn）的设置通常使用框架的服务发现功能。
除非设置要求使用特定的端口范围或特定的网络接口进行绑定，否则不需要配置任何TaskManager的主机和端口。
| 键 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| jobmanager.rpc.address | (none) | 字符串 | 定义与作业管理器通信的网络地址的配置参数。该值仅在存在具有静态名称或地址的单个JobManager的设置中解释（简单的独立设置或具有动态服务名称解析的容器设置）。在许多高可用性设置中，不使用此选项，而是使用领导者选举服务（如ZooKeeper）从可能存在的多个备用JobManager中选举和发现JobManager领导者。 |
| jobmanager.rpc.port | 6123 | 整数 | 定义与作业管理器通信的网络端口的配置参数。与jobmanager.rpc.address一样，该值仅在存在具有静态名称/地址和端口的单个JobManager的设置中解释（简单的独立设置或具有动态服务名称解析的容器设置）。此配置选项在许多高可用性设置中不使用，而是使用领导者选举服务（如ZooKeeper）从可能存在的多个备用JobManager中选举和发现JobManager领导者。 |
| metrics.internal.query-service.port | “0” | 字符串 | Flink内部度量查询服务使用的端口范围。接受端口列表（“50100,50101”）、范围（“50100-50200”）或两者的组合。建议设置一系列端口以避免多个Flink组件在同一台机器上运行时发生冲突。默认情况下，Flink将选择一个随机端口。 |
| rest.address | (none) | 字符串 | 客户端用于连接到服务器的地址。注意：仅在高可用性配置为NONE时才会考虑此选项。 |
| rest.bind-address | (none) | 字符串 | 服务器绑定的地址。 |
| rest.bind-port | “8081” | 字符串 | 服务器绑定的端口。接受端口列表（“50100,50101”）、范围（“50100-50200”）或两者的组合。建议设置一系列端口以避免多个Rest服务器在同一台机器上运行时发生冲突。 |
| rest.port | 8081 | 整数 | 客户端连接的端口。如果未指定rest.bind-port，则REST服务器将绑定到此端口。注意：仅当高可用性配置为NONE时才会考虑此选项。 |
| taskmanager.data.port | 0 | 整数 | 任务管理器用于数据交换操作的外部端口。 |
| taskmanager.host | (none) | 字符串 | TaskManager暴露的网络接口的外部地址。由于不同的TaskManagers需要不同的值，通常在额外的非共享的TaskManager-specific配置文件中指定。 |
| taskmanager.rpc.port | “0” | 字符串 | TaskManager暴露的外部RPC端口。接受端口列表（“50100,50101”）、范围（“50100-50200”）或两者的组合。建议设置一系列端口以避免多个TaskManager在同一台机器上运行时发生冲突。 |
容错
这些配置选项控制Flink在执行过程中出现故障时的重启行为。通过在您的flink-conf.yaml中配置这些选项，您可以定义集群的默认重启策略。
如果没有通过ExecutionConfig配置特定作业的重启策略，则默认重启策略将生效。
| 键 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| restart-strategy.type | (none) | 字符串 | 定义在作业失败时使用的重启策略。接受的值有：none，off，disable：无重启策略。fixeddelay，fixed-delay：固定延迟重启策略。更多详情请参见这里。failurerate，failure-rate：故障率重启策略。更多详情请参见这里。exponentialdelay，exponential-delay：指数延迟重启策略。更多详情请参见这里。如果禁用了检查点，default值是none。如果启用了检查点，default值是具有Integer.MAX_VALUE重启尝试和"1 s"延迟的fixed-delay。 |
固定延迟重启策略
| 键 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| restart-strategy.fixed-delay.attempts | 1 | 整数 | 如果restart-strategy.type被设置为fixed-delay，则Flink在作业被声明为失败之前重试执行的次数。 |
| restart-strategy.fixed-delay.delay | 1 s | 持续时间 | 如果restart-strategy.type被设置为fixed-delay，则是两次连续重启尝试之间的延迟。当程序与外部系统交互时，例如在重新执行之前连接或未决事务应达到超时。可以使用以下表示法指定：“1 min”，“20 s” |
故障率重启策略
| 键 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| restart-strategy.failure-rate.delay | 1 s | 持续时间 | 如果restart-strategy.type被设置为failure-rate，则是两次连续重启尝试之间的延迟。可以使用以下表示法指定：“1 min”，“20 s” |
| restart-strategy.failure-rate.failure-rate-interval | 1 min | 持续时间 | 如果restart-strategy.type被设置为failure-rate，则测量故障率的时间间隔。可以使用以下表示法指定：“1 min”，“20 s” |
| restart-strategy.failure-rate.max-failures-per-interval | 1 | 整数 | 如果restart-strategy.type被设置为failure-rate，则在给定时间间隔内的最大重启次数，否则将失败作业。 |
可重试的清理
作业达到全局终止状态后，将执行所有相关资源的清理操作。如果出现故障，此清理可以进行重试。可以配置不同的重试策略来更改此行为：
| 键 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| cleanup-strategy.type | “exponential-delay” | 字符串 | 定义在清理失败时使用的清理策略。接受的值有：none，disable，off：仅执行一次清理操作。如果发生故障，将不会启动重试。在故障的情况下，作业的工件（以及作业的JobResultStore条目）需要手动清理。fixed-delay，fixeddelay：在固定间隔内尝试清理，直到清理被视为成功或达到一组重试次数。达到配置的限制意味着作业的工件（以及作业的JobResultStore条目）可能需要手动清理。exponential-delay，exponentialdelay：指数延迟重启策略触发清理，延迟以指数增长，直到清理成功或达到一组重试次数。达到配置的限制意味着作业的工件（以及作业的JobResultStore条目）可能需要手动清理。默认配置依赖于具有给定默认值的指数延迟重试策略。 |
固定延迟清理重试策略
| 键 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| cleanup-strategy.fixed-delay.attempts | infinite | 整数 | 如果cleanup-strategy.type被设置为fixed-delay，则Flink在放弃之前重试清理的次数。达到配置的限制意味着作业的工件（以及作业的JobResultStore条目）可能需要手动清理。 |
| cleanup-strategy.fixed-delay.delay | 1 min | 持续时间 | 如果cleanup-strategy.type被设置为fixed-delay，则是两次连续重试尝试之间的延迟。可以使用以下表示法指定：“1 min”，“20 s” |
指数延迟清理重试策略
| 键 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| cleanup-strategy.exponential-delay.attempts | infinite | 整数 | 如果cleanup-strategy.type被设置为exponential-delay，则在清理失败的情况下重试的次数。达到配置的限制意味着作业的工件（以及作业的JobResultStore条目）可能需要手动清理。 |
| cleanup-strategy.exponential-delay.initial-backoff | 1 s | 持续时间 | 如果cleanup-strategy.type被设置为exponential-delay，则是两个连续清理尝试之间的初始延迟。可以使用以下表示法指定：“1 min”，“20 s” |
| cleanup-strategy.exponential-delay.max-backoff | 1 h | 持续时间 | 如果cleanup-strategy.type被设置为exponential-delay，则是两个连续清理尝试之间的最长延迟。可以使用以下表示法指定：“1 min”，“20 s” |
检查点和状态后端
这些选项控制状态后端和检查点行为的基本设置。
这些选项仅适用于以连续流方式执行的作业/应用程序。以批处理方式执行的作业/应用程序不使用状态后端和检查点，而是使用针对批处理处理进行了优化的不同内部数据结构。
| 键 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| state.backend.type | (none) | String | 用于存储状态的状态后端。可以通过其快捷名称或StateBackendFactory的类名指定实现。如果指定了工厂，则通过其零参数构造函数进行实例化，并调用其StateBackendFactory#createFromConfig(ReadableConfig, ClassLoader)方法。识别的快捷名称为’hashmap’和’rocksdb’。 |
| state.checkpoint-storage | (none) | String | 用于检查点状态的检查点存储实现。可以通过其快捷名称或CheckpointStorageFactory的类名指定实现。如果指定了工厂，则通过其零参数构造函数进行实例化，并调用其CheckpointStorageFactory#createFromConfig(ReadableConfig, ClassLoader)方法。识别的快捷名称为’jobmanager’和’filesystem’。 |
| state.checkpoints.dir | (none) | String | 用于在Flink支持的文件系统中存储检查点的数据文件和元数据的默认目录。存储路径必须对所有参与进程/节点（即所有TaskManager和JobManager）都是可访问的。 |
| state.savepoints.dir | (none) | String | 保存点的默认目录。由将保存点写入文件系统的状态后端（HashMapStateBackend、EmbeddedRocksDBStateBackend）使用。 |
| state.backend.incremental | false | Boolean | 选项，指定状态后端是否应创建增量检查点（如果可能）。对于增量检查点，仅存储与上一个检查点的差异，而不是完整的检查点状态。启用后，Web UI中显示的状态大小或从REST API获取的状态大小仅表示增量检查点大小，而不是完整的检查点大小。某些状态后端可能不支持增量检查点并忽略此选项。 |
| state.backend.local-recovery | false | Boolean | 此选项为该状态后端配置本地恢复。默认情况下，禁用本地恢复。本地恢复目前仅覆盖键控状态后端（包括EmbeddedRocksDBStateBackend和HashMapStateBackend）。 |
| state.checkpoints.num-retained | 1 | Integer | 要保留的完成检查点的最大数量。 |
| taskmanager.state.local.root-dirs | (none) | String | 定义用于存储用于本地恢复的基于文件的状态的根目录的配置参数。本地恢复目前仅覆盖键控状态后端。如果未配置，则默认为<WORKING_DIR>/localState。可以通过process.taskmanager.working-dir配置<WORKING_DIR>。 |
高可用性
此处的高可用性是指JobManager进程从故障中恢复的能力。
JobManager确保在TaskManagers之间进行恢复期间的一致性。为了使JobManager自身能够一致地恢复，外部服务必须存储最小量的恢复元数据（如“上次提交的检查点的ID”），并帮助选择和锁定哪个JobManager是领导者（以避免分裂大脑情况）。
| 键 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| high-availability.type | “NONE” | String | 定义用于集群执行的高可用性模式。要启用高可用性，请将此模式设置为“ZOOKEEPER”，“KUBERNETES”或指定完全限定名称的工厂类。 |
| high-availability.cluster-id | “/default” | String | Flink集群的ID，用于将多个Flink集群彼此分离。对于独立集群，需要设置该值，但在YARN中会自动推断。 |
| high-availability.storageDir | (none) | String | Flink在高可用性设置中持久化元数据的文件系统路径（URI）。 |
高可用性设置中JobResultStore的选项
| 键 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| job-result-store.delete-on-commit | true | Boolean | 确定是否在相应实体转换为干净状态时自动从底层作业结果存储中删除作业结果。如果为false，则将清理的作业结果标记为干净状态以指示其状态。在这种情况下，Flink不再拥有所有权，并且资源需要由用户进行清理。 |
| job-result-store.storage-path | (none) | String | 定义作业结果应存储的位置。这应该是提供读写一致性的底层文件系统。默认情况下，这是{high-availability.storageDir}/job-result-store/{high-availability.cluster-id}。 |
与ZooKeeper的高可用性设置选项
| 键 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| high-availability.zookeeper.path.root | “/flink” | String | Flink在ZooKeeper中存储其条目的根路径。 |
| high-availability.zookeeper.quorum | (none) | String | 在使用ZooKeeper的高可用性模式下运行Flink时要使用的ZooKeeper群集。 |
内存配置
这些配置值控制TaskManagers和JobManagers使用内存的方式。
Flink尽可能地屏蔽用户对数据密集处理的JVM进行配置的复杂性。在大多数情况下，用户只需要设置taskmanager.memory.process.size或taskmanager.memory.flink.size（取决于设置方式），并可能通过taskmanager.memory.managed.fraction调整JVM堆和管理内存的比例。下面的其他选项可用于性能调优和修复与内存相关的错误。
有关这些选项如何相互作用的详细说明，请参阅TaskManager和JobManager内存配置的文档。
| 键 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| jobmanager.memory.enable-jvm-direct-memory-limit | false | Boolean | 是否启用JobManager进程的JVM直接内存限制（-XX:MaxDirectMemorySize）。限制将设置为’jobmanager.memory.off-heap.size’选项的值。 |
| jobmanager.memory.flink.size | (none) | MemorySize | JobManager的总Flink内存大小。这包括JobManager消耗的所有内存，除了JVM Metaspace和JVM Overhead之外。它由JVM堆内存和堆外内存组成。有关总进程内存大小配置的详细信息，请参见’jobmanager.memory.process.size’。 |
| jobmanager.memory.heap.size | (none) | MemorySize | JobManager的JVM堆内存大小。最小推荐的JVM堆大小为128.000mb（134217728字节）。 |
| jobmanager.memory.jvm-metaspace.size | 256 mb | MemorySize | JobManager的JVM Metaspace大小。 |
| jobmanager.memory.jvm-overhead.fraction | 0.1 | Float | 用于JVM Overhead的Total Process Memory的比例。这是为JVM开销保留的堆外内存，例如线程堆栈空间，编译缓存等。这包括本机内存但不包括直接内存，并且在Flink计算JVM max direct memory size参数时不会计入其中。JVM Overhead的大小是通过将Total Process Memory的配置分数补足而派生的。如果派生大小小于或大于配置的最小或最大大小，则将使用最小或最大大小。可以通过将最小值和最大值设置为相同值来显式指定JVM Overhead的确切大小。 |
| jobmanager.memory.jvm-overhead.max | 1 gb | MemorySize | JobManager的最大JVM Overhead大小。这是为JVM开销保留的堆外内存，例如线程堆栈空间，编译缓存等。这包括本机内存但不包括直接内存，并且在Flink计算JVM max direct memory size参数时不会计入其中。JVM Overhead的大小是通过将Total Process Memory的配置分数补足而派生的。如果派生大小小于或大于配置的最小或最大大小，则将使用最小或最大大小。可以通过将最小值和最大值设置为相同值来显式指定JVM Overhead的确切大小。 |
| jobmanager.memory.jvm-overhead.min | 192 mb | MemorySize | JobManager的最小JVM Overhead大小。这是为JVM开销保留的堆外内存，例如线程堆栈空间，编译缓存等。这包括本机内存但不包括直接内存，并且在Flink计算JVM max direct memory size参数时不会计入其中。JVM Overhead的大小是通过将Total Process Memory的配置分数补足而派生的。如果派生大小小于或大于配置的最小或最大大小，则将使用最小或最大大小。可以通过将最小值和最大值设置为相同值来显式指定JVM Overhead的确切大小。 |
| jobmanager.memory.off-heap.size | 128 mb | MemorySize | JobManager的堆外内存大小。此选项涵盖所有堆外内存使用，包括直接内存和本机内存分配。如果启用了’jobmanager.memory.enable-jvm-direct-memory-limit’，则JobManager进程的JVM直接内存限制（-XX:MaxDirectMemorySize）将设置为此值。 |
| jobmanager.memory.process.size | (none) | MemorySize | JobManager的总进程内存大小。这包括JobManager JVM进程消耗的所有内存，包括总Flink内存、JVM Metaspace和JVM Overhead。在容器化设置中，应将其设置为容器内存。有关Total Flink Memory大小配置的详细信息，请参见’jobmanager.memory.flink.size’。 |
| taskmanager.memory.flink.size | (none) | MemorySize | TaskExecutor的总Flink内存大小。这包括TaskExecutor消耗的所有内存，除了JVM Metaspace和JVM Overhead之外。它由Framework Heap Memory、Task Heap Memory、Task Off-Heap Memory、Managed Memory和Network Memory组成。有关总进程内存大小配置的详细信息，请参见’taskmanager.memory.process.size’。 |
| taskmanager.memory.framework.heap.size | 128 mb | MemorySize | TaskExecutor框架的堆内存大小。这是为TaskExecutor框架保留的JVM堆内存大小，不会分配给任务槽。 |
| taskmanager.memory.framework.off-heap.batch-shuffle.size | 64 mb | MemorySize | 批量洗牌使用的内存大小，用于读取洗牌数据（目前仅由排序洗牌和混合洗牌使用）。注意：该内存从’taskmanager.memory.framework.off-heap.size’中减少，因此必须小于该值，这意味着您可能还需要增加此配置值后增加’taskmanager.memory.framework.off-heap.size’；此内存大小可能会影响洗牌性能，可以为大规模批处理作业增加此配置值（例如128M或256M）。 |
| taskmanager.memory.framework.off-heap.size | 128 mb | MemorySize | TaskExecutor框架的堆外内存大小。这是为TaskExecutor框架保留的堆外内存（JVM直接内存和本机内存）的大小，不会分配给任务槽。在Flink计算JVM max direct memory size参数时，将完全计入配置的值。 |
| taskmanager.memory.jvm-metaspace.size | 256 mb | MemorySize | TaskExecutor的JVM Metaspace大小。 |
| taskmanager.memory.jvm-overhead.fraction | 0.1 | Float | 用于JVM Overhead的Total Process Memory的比例。这是为JVM开销保留的堆外内存，例如线程堆栈空间，编译缓存等。这包括本机内存但不包括直接内存，并且在Flink计算JVM max direct memory size参数时不会计入其中。JVM Overhead的大小是通过将Total Process Memory的配置分数补足而派生的。如果派生大小小于或大于配置的最小或最大大小，则将使用最小或最大大小。可以通过将最小值和最大值设置为相同值来显式指定JVM Overhead的确切大小。 |
| taskmanager.memory.jvm-overhead.max | 1 gb | MemorySize | TaskExecutor的最大JVM Overhead大小。这是为JVM开销保留的堆外内存，例如线程堆栈空间，编译缓存等。这包括本机内存但不包括直接内存，并且在Flink计算JVM max direct memory size参数时不会计入其中。JVM Overhead的大小是通过将Total Process Memory的配置分数补足而派生的。如果派生大小小于或大于配置的最小或最大大小，则将使用最小或最大大小。可以通过将最小值和最大值设置为相同值来显式指定JVM Overhead的确切大小。 |
| taskmanager.memory.jvm-overhead.min | 192 mb | MemorySize | TaskExecutor的最小JVM Overhead大小。这是为JVM开销保留的堆外内存，例如线程堆栈空间，编译缓存等。这包括本机内存但不包括直接内存，并且在Flink计算JVM max direct memory size参数时不会计入其中。JVM Overhead的大小是通过将Total Process Memory的配置分数补足而派生的。如果派生大小小于或大于配置的最小或最大大小，则将使用最小或最大大小。可以通过将最小值和最大值设置为相同值来显式指定JVM Overhead的确切大小。 |
| taskmanager.memory.managed.consumer-weights | OPERATOR:70,STATE_BACKEND:70,PYTHON:30 | Map | 不同类型消费者的托管内存权重。一个槽的托管内存由其包含的所有类型的消费者共享，与每种类型的消费者数量无关，比例与权重成正比。当前支持的消费者类型为OPERATOR（用于内置算法）、STATE_BACKEND（用于RocksDB状态后端）和PYTHON（用于Python进程）。 |
| taskmanager.memory.managed.fraction | 0.4 | Float | 要作为托管内存使用的Total Flink Memory的比例，如果未明确指定托管内存大小。 |
| taskmanager.memory.managed.size | (none) | MemorySize | TaskExecutor的托管内存大小。这是由内存管理器管理的堆外内存的大小，用于排序、哈希表、缓存中间结果和RocksDB状态后端。内存消费者可以通过从内存管理器分配MemorySegments的形式分配内存，或者从内存管理器保留字节并将其内存使用保持在该边界内。如果未指定，则会派生出配置分数补足Total Flink Memory的大小。 |
| taskmanager.memory.network.fraction | 0.1 | Float | 要用作网络内存的Total Flink Memory的比例。网络内存是为ShuffleEnvironment（例如网络缓冲区）保留的堆外内存。网络内存的大小是通过将Total Flink Memory的配置分数补足而派生的。如果派生大小小于或大于配置的最小或最大大小，则将使用最小或最大大小。可以通过将最小值和最大值设置为相同值来显式指定网络内存的确切大小。 |
| taskmanager.memory.network.max | infinite | MemorySize | TaskExecutor的最大网络内存大小。网络内存是为ShuffleEnvironment（例如网络缓冲区）保留的堆外内存。网络内存的大小是通过将Total Flink Memory的配置分数补足而派生的。如果派生大小小于或大于配置的最小或最大大小，则将使用最小或最大大小。默认情况下，Network Memory的最大限制为Long.MAX_VALUE。可以通过将最小/最大值设置为相同值来显式指定网络内存的确切大小。 |
| taskmanager.memory.network.min | 64 mb | MemorySize | TaskExecutor的最小网络内存大小。网络内存是为ShuffleEnvironment（例如网络缓冲区）保留的堆外内存。网络内存的大小是通过将Total Flink Memory的配置分数补足而派生的。如果派生大小小于或大于配置的最小或最大大小，则将使用最小或最大大小。可以通过将最小值和最大值设置为相同值来显式指定网络内存的确切大小。 |
| taskmanager.memory.process.size | (none) | MemorySize | TaskExecutor的总进程内存大小。这包括TaskExecutor消耗的所有内存，包括Total Flink Memory、JVM Metaspace和JVM Overhead。在容器化设置中，应将其设置为容器内存。有关Total Flink Memory大小配置的详细信息，请参见’taskmanager.memory.flink.size’。 |
| taskmanager.memory.task.heap.size | (none) | MemorySize | TaskExecutor的任务堆内存大小。这是为任务保留的JVM堆内存大小。如果未指定，它将派生为Total Flink Memory减去Framework Heap Memory、Framework Off-Heap Memory、Task Off-Heap Memory、Managed Memory和Network Memory。 |
| taskmanager.memory.task.off-heap.size | 0 bytes | MemorySize | TaskExecutor的任务堆外内存大小。这是为任务保留的堆外内存（JVM直接内存和本机内存）的大小。在Flink计算JVM max direct memory size参数时，将完全计入配置的值。 |
杂项选项
| 键 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| fs.allowed-fallback-filesystems | (none) | String | 文件方案的（分号分隔的）列表，可以在适当的Flink插件之外使用Hadoop代替。 （例如：s3;wasb） |
| fs.default-scheme | (none) | String | 默认文件系统方案，用于未明确声明方案的路径。可以包含权限，例如HDFS NameNode的主机：端口。 |
| io.tmp.dirs | Yarn上的’LOCAL_DIRS’。独立环境中为System.getProperty(“java.io.tmpdir”) | String | 临时文件的目录，由逗号“，”、“ |
安全性
配置Flink的安全性和与外部系统的安全交互选项。
SSL
通过SSL可以保护Flink的网络连接。请参阅SSL设置文档，了解详细的设置指南和背景信息。
| 键名 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| security.ssl.algorithms | “TLS_RSA_WITH_AES_128_CBC_SHA” | 字符串 | 支持的标准SSL算法的逗号分隔列表。在这里阅读更多详情 |
| security.ssl.internal.cert.fingerprint | (无) | 字符串 | 内部证书的SHA1指纹。进一步保护内部通信以呈现Flink使用的确切证书。在无法使用私有CA（自签名）或需要内部公司级CA的情况下，这是必需的。 |
| security.ssl.internal.enabled | false | 布尔值 | 启用内部网络通信的SSL。可选地，特定组件可以通过其自己的设置（RPC、数据传输、REST等）覆盖此设置。 |
| security.ssl.internal.key-password | (无) | 字符串 | 解密Flink内部端点（RPC、数据传输、Blob服务器）的密钥的秘密。 |
| security.ssl.internal.keystore | (无) | 字符串 | 带有SSL密钥和证书的Java密钥库文件，用于Flink的内部端点（RPC、数据传输、Blob服务器）。 |
| security.ssl.internal.keystore-password | (无) | 字符串 | 解密Flink的内部端点（RPC、数据传输、Blob服务器）的密钥库文件的秘密。 |
| security.ssl.internal.truststore | (无) | 字符串 | 包含公共CA证书以验证Flink的内部端点（RPC、数据传输、Blob服务器）对等方的信任库文件。 |
| security.ssl.internal.truststore-password | (无) | 字符串 | 解密Flink的内部端点（RPC、数据传输、Blob服务器）的信任库文件的密码。 |
| security.ssl.protocol | “TLSv1.2” | 字符串 | SSL传输支持的SSL协议版本。请注意，它不支持逗号分隔列表。 |
| security.ssl.rest.authentication-enabled | false | 布尔值 | 通过REST端点启用外部通信的互相SSL身份验证。 |
| security.ssl.rest.cert.fingerprint | (无) | 字符串 | rest证书的SHA1指纹。进一步保护rest REST端点，以呈现仅由代理服务器使用的证书。这在使用公共CA或内部公司级CA时是必需的。 |
| security.ssl.rest.enabled | false | 布尔值 | 启用通过REST端点的外部通信的SSL。 |
| security.ssl.rest.key-password | (无) | 字符串 | 解密Flink的外部REST端点中的密钥的秘密。 |
| security.ssl.rest.keystore | (无) | 字符串 | 带有SSL密钥和证书的Java密钥库文件，用于Flink的外部REST端点。 |
| security.ssl.rest.keystore-password | (无) | 字符串 | 解密Flink的外部REST端点的密钥库文件的秘密。 |
| security.ssl.rest.truststore | (无) | 字符串 | 包含公共CA证书以验证Flink的外部REST端点对等方的信任库文件。 |
| security.ssl.rest.truststore-password | (无) | 字符串 | 解密Flink的外部REST端点的信任库文件的密码。 |
| security.ssl.verify-hostname | true | 布尔值 | 在SSL握手期间启用对对等方主机名的验证的标志。 |
与外部系统进行身份验证
委派令牌
Flink具有可插拔的身份验证协议不可知的委派令牌框架。请参阅Flink和委派令牌文档了解详细信息。
| 键名 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| security.delegation.tokens.enabled | true | 布尔值 | 是否启动外部服务的委派令牌系统。 |
| security.delegation.tokens.renewal.retry.backoff | 1 h | 持续时间 | 失败后重新获取新的委派令牌之前等待的时间段。 |
| security.delegation.tokens.renewal.time-ratio | 0.75 | 双精度数 | 在何时重新获取新凭证时，令牌到期时间的比率。 |
| security.delegation.token.provider..enabled | true | 布尔值 | 控制是否在启用安全性时为服务获取凭证。默认情况下，配置了这些服务的所有支持的凭证都会被检索，但如果与正在运行的应用程序冲突，可以禁用该行为。 |
ZooKeeper身份验证/授权
当连接到受保护的ZooKeeper quorum时，需要这些选项。
| 键名 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| zookeeper.sasl.disable | false | 布尔值 | |
| zookeeper.sasl.login-context-name | “Client” | 字符串 | |
| zookeeper.sasl.service-name | “zookeeper” | 字符串 |
基于Kerberos的身份验证/授权
请参阅Flink和Kerberos文档以获取设置指南和Flink可以通过Kerberos验证自身的外部系统列表。
| 键名 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| security.kerberos.access.hadoopFileSystems | (无) | List | Kerberos保护的Hadoop文件系统的分号分隔列表。例如，security.kerberos.access.hadoopFileSystems=hdfs://namenode2:9002;hdfs://namenode3:9003。JobManager需要访问这些文件系统以检索安全令牌。 |
| security.kerberos.login.contexts | (无) | 字符串 | 提供Kerberos凭据的逗号分隔的登录上下文列表（例如，“Client，KafkaClient”以使用ZooKeeper身份验证和Kafka身份验证的凭据） |
| security.kerberos.login.keytab | (无) | 字符串 | 包含用户凭据的Kerberos密钥表文件的绝对路径。 |
| security.kerberos.login.principal | (无) | 字符串 | 与密钥表关联的Kerberos主体名称。 |
| security.kerberos.login.use-ticket-cache | true | 布尔值 | 指示是否从Kerberos票据缓存中读取。 |
| security.kerberos.relogin.period | 1 min | 持续时间 | 定期自动发生的关键登录的时间段，以始终拥有有效的TGT。 |
资源编排框架
本节包含与将Flink与资源编排框架（如Kubernetes、Yarn等）集成相关的选项，但并非总是必要的。例如，您可以在不让Flink知道它在Kubernetes上运行（而且不指定这里的任何Kubernetes配置选项）的情况下轻松部署Flink应用程序。请参阅此设置指南作为示例。
这些选项适用于Flink本身主动从编排器请求和释放资源的设置。
YARN
| 键名 | 默认值 | 类型 | 描述 |
|---|---|---|---|
| external-resource.<resource_name>.yarn.config-key | (无) | 字符串 | 如果配置，Flink将向容器请求的资源配置文件添加此键。该值将设置为external-resource.<resource_name>.amount的值。 |
| flink.hadoop. | (无) | 字符串 | 通过前缀’flink.hadoop.'探测Hadoop配置的通用选项。Flink将删除前缀以获取（来自core-default.xml和hdfs-default.xml），然后将和值设置为Hadoop配置。例如，在Flink配置中设置flink.hadoop.dfs.replication=5，并在Hadoop配置中转换为dfs.replication=5。 |
| flink.yarn. | (无) | 字符串 | 通过前缀’flink.yarn.'探测Yarn配置的通用选项。Flink将删除前缀’flink.'以获取yarn.（来自yarn-default.xml），然后将yarn.和值设置为Yarn配置。例如，在Flink配置中设置flink.yarn.resourcemanager.container.liveness-monitor.interval-ms=300000，并在Yarn配置中转换为yarn.resourcemanager.container.liveness-monitor.interval-ms=300000。 |
| yarn.application-attempt-failures-validity-interval | 10000 | 长整型 | 定义在重新启动AM时，应用程序尝试失败的次数的时间窗口（以毫秒为单位）。在此窗口之外的故障不予考虑。将此值设置为-1以全局计算。有关详细信息，请参见此处。 |
| yarn.application-attempts | (无) | 字符串 | ApplicationMaster重启次数。默认情况下，该值将设置为1。如果启用了高可用性，则默认值将设置为2。重新启动次数还受YARN的限制（通过yarn.resourcemanager.am.max-attempts进行配置）。请注意，整个Flink集群将重新启动，并且YARN客户端将丢失连接。 |
| yarn.application-master.port | “0” | 字符串 | 使用此配置选项，用户可以为应用程序主节点（和JobManager）RPC端口指定一个端口、一系列端口或端口列表。我们建议使用默认值（0），让操作系统选择适当的端口。特别是当多个AM在同一物理主机上运行时，固定端口分配会阻止AM启动。例如，在具有限制性防火墙的环境中在YARN上运行Flink时，此选项允许指定允许的端口范围。 |
| yarn.application.id | (无) | 字符串 | 运行的YARN应用程序的YARN应用程序ID。这是将执行管道的YARN群集。 |
| yarn.application.name | (无) | 字符串 | YARN应用程序的自定义名称。 |
| yarn.application.node-label | (无) | 字符串 | 为YARN应用程序指定YARN节点标签。 |
| yarn.application.priority | -1 | 整型 | 表示提交Flink YARN应用程序的优先级的非负整数。仅在启用了YARN优先级调度设置时才会生效。较大的整数对应更高的优先级。如果优先级为负数或设置为“-1”（默认值），则Flink将取消yarn优先级设置并使用集群默认优先级。有关启用目标YARN版本的优先级调度所需的特定设置，请参阅YARN的官方文档。 |
| yarn.application.queue | (无) | 字符串 | 要放置当前管道的YARN队列。 |
| yarn.application.type | (无) | 字符串 | YARN应用程序的自定义类型。 |
| yarn.appmaster.vcores | 1 | 整型 | YARN应用程序主节点使用的虚拟核心（vcore）数量。 |
| yarn.classpath.include-user-jar | ORDER | 枚举 | 定义是否将用户jar包包含在系统类路径中以及它们在路径中的位置。可选值：“DISABLED”：将用户jar从系统类路径中排除“FIRST”：放在开头“LAST”：放在末尾“ORDER”：基于jar的名称进行排序 |
| yarn.containers.vcores | -1 | 整型 | 每个YARN容器使用的虚拟核心（vcore）数。默认情况下，如果设置了TaskManager的插槽数量，则将vcore数设置为插槽数量；否则设置为1。为了使用此参数，您的集群必须启用CPU调度。您可以通过设置org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler来实现这一点。 |
| yarn.file-replication | -1 | 整型 | 每个本地资源文件的文件复制数量。如果未配置，则Flink将使用hadoop配置中的默认复制值。 |
| yarn.flink-dist-jar | (无) | 字符串 | Flink dist jar的位置。 |
| yarn.heartbeat.container-request-interval | 500 | 整型 | 在毫秒中ResourceManager和Flink请求容器之间的心跳时间：这个值越低，Flink会尽快收到有关容器分配的通知，因为请求和分配是通过心跳传输的。这个值越低，可能会分配更多的多余容器，但会对Yarn施加压力。如果观察到ResourceManager上有太多的容器分配，请增加此值。有关更多信息，请参见此链接。 |
| yarn.heartbeat.interval | 5 | 整型 | 与ResourceManager之间的心跳间隔，以秒为单位。 |
| yarn.properties-file.location | (无) | 字符串 | 当将Flink作业提交到YARN时，JobManager的主机和可用处理插槽数会写入一个属性文件中，以便Flink客户端能够获取这些详细信息。此配置参数允许更改该文件的默认位置（例如，对于在用户之间共享Flink安装的环境）。 |
| yarn.provided.lib.dirs | (无) | List | 提供的库目录的分号分隔列表。它们应该事先上传并且可由全局读取。Flink将使用它们来排除本地Flink jars（如flink-dist、lib/、plugins/）的上传以加快作业提交过程。此外，YARN将在节点上缓存它们，因此不需要每次应用程序都下载它们。一个示例可以是hdfs://$namenode_address/path/of/flink/lib |
| yarn.provided.usrlib.dir | (无) | 字符串 | 远程提供的usrlib目录。它应事先上传并且对全球读取。Flink将使用它来排除本地usrlib目录（即FLINK_LIB_DIR的父目录下的usrlib/）。与yarn.provided.lib.dirs不同，YARN不会将其缓存在节点上，因为它适用于每个应用程序。一个示例可以是hdfs://$namenode_address/path/of/flink/usrlib |
| yarn.security.kerberos.localized-keytab-path | “krb5.keytab” | 字符串 | Kerberos密钥表文件将本地化到的路径（在NodeManager上）。如果yarn.security.kerberos.ship-local-keytab设置为true，Flink将作为YARN本地资源发送keytab文件。在这种情况下，该路径是相对于本地资源目录。如果设置为false，Flink将尝试直接从路径自身定位keytab。 |
| yarn.security.kerberos.ship-local-keytab | true | 布尔值 | 当此值为true时，Flink将通过security.kerberos.login.keytab配置的keytab文件作为本地化的YARN资源发送。 |
| yarn.ship-archives | (无) | List | 要提供给YARN集群的存档的分号分隔列表。当本地化时，这些存档将被解压缩，并且可以是以下类型之一：“.tar.gz”、“.tar”、“.tgz”、“.dst”、“.jar”、“.zip”。 |
| yarn.ship-files | (无) | List | 要提供给YARN集群的文件和/或目录的分号分隔列表。 |
| yarn.staging-directory | (无) | 字符串 | 提交应用程序时用于存储YARN文件的暂存目录。默认情况下，它使用配置的文件系统的主目录。 |
| yarn.tags | (无) | 字符串 | 要应用于Flink YARN应用程序的标签的逗号分隔列表。 |
| yarn.taskmanager.node-label | (无) | 字符串 | 为Flink任务管理器指定的YARN节点标签，它将覆盖TaskManagers的yarn.application.node-label，如果两者都设置了。 |
Kubernetes
Kubernetes是一个开源的容器编排平台，用于自动化部署、扩展和管理容器化应用程序。它提供了一种简单且高效的方式来管理容器，使得在大规模集群上运行应用变得更加容易。
以下是一些与Kubernetes相关的配置选项：
| Key | Default | Type | Description |
|---|---|---|---|
| external-resource.<resource_name>.kubernetes.config-key | (none) | String | 如果配置了此项，Flink将向TaskExecutor的主容器中添加"resources.limits.“和"resources.requests.”，并将值设置为external-resource.<resource_name>.amount的值。 |
| kubernetes.client.io-pool.size | 4 | Integer | Kubernetes客户端使用的IO执行器池的大小，用于执行阻塞IO操作（例如启动/停止TaskManager pod，更新与领导者相关的ConfigMaps等）。增加池的大小可以同时运行更多的IO操作。 |
| kubernetes.client.user-agent | “flink” | String | 用于与Kubernetes API服务器通信的用户代理。 |
| kubernetes.cluster-id | (none) | String | 集群ID，用于标识唯一的Flink集群。ID只能包含小写字母数字字符和"-“，必须满足格式"a-z”。如果未设置，客户端将自动生成一个随机ID。 |
| kubernetes.config.file | (none) | String | 用于创建Kubernetes客户端的Kubernetes配置文件。默认位于~/.kube/config |
| kubernetes.container.image.pull-policy | IfNotPresent | Enum | Kubernetes容器镜像拉取策略。默认策略为IfNotPresent，以避免对镜像仓库施加压力。可能的值：“IfNotPresent”、“Always”、“Never” |
| kubernetes.container.image.pull-secrets | (none) | List | 用于访问私有镜像注册表的Kubernetes密钥的分号分隔列表。 |
| kubernetes.container.image.ref | 默认值取决于实际运行的版本。通常为"flink:<FLINK_VERSION>-scala_<SCALA_VERSION>" | String | 用于Flink容器的镜像。指定的镜像必须基于与应用程序使用的Apache Flink和Scala版本相同的版本。请访问此处提供的Flink项目官方Docker镜像链接。Flink项目还会发布docker镜像到apache/flink DockerHub仓库。 |
| kubernetes.context | (none) | String | 要在Kubernetes配置文件中使用的所需上下文，用于配置与集群交互的Kubernetes客户端。如果用户配置了多个上下文，并希望在不同的Kubernetes集群/上下文上管理不同的Flink集群，这将非常有用。 |
| kubernetes.decorator.hadoop-conf-mount.enabled | true | Boolean | 是否启用Hadoop配置挂载修饰器。当Hadoop配置在Flink之外挂载时，必须将其设置为false。一个典型的用例是使用Flink Kubernetes Operator。 |
| kubernetes.decorator.kerberos-mount.enabled | true | Boolean | 是否启用Kerberos挂载修饰器。当Kerberos配置和keytab在Flink之外挂载时，必须将其设置为false。一个典型的用例是使用Flink Kubernetes Operator。 |
| kubernetes.entry.path | “/docker-entrypoint.sh” | String | 镜像中Kubernetes的入口脚本，将用作jobmanager和taskmanager容器的命令。 |
| kubernetes.env.secretKeyRef | (none) | List | 用户指定的将env变量设置为Flink容器中的密钥引用。值应采用以下形式：env:FOO_ENV,secret:foo_secret,key:foo_key;env:BAR_ENV,secret:bar_secret,key:bar_key。 |
| kubernetes.flink.conf.dir | “/opt/flink/conf” | String | 将在pod中挂载的flink conf目录。该路径中的flink-conf.yaml、log4j.properties和logback.xml将从配置映射中覆盖。 |
| kubernetes.flink.log.dir | (none) | String | 保存jobmanager和taskmanager日志的目录。默认值为$FLINK_HOME/log。 |
| kubernetes.hadoop.conf.config-map.name | (none) | String | 指定包含要挂载到JobManager和TaskManagers上的自定义Hadoop配置的现有ConfigMap的名称。 |
| kubernetes.hostnetwork.enabled | false | Boolean | 是否启用HostNetwork模式。HostNetwork允许pod使用节点网络命名空间而不是单独的pod网络命名空间。请注意，JobManager服务帐户应具有更新Kubernetes服务的权限。 |
| kubernetes.jobmanager.annotations | (none) | Map | 要设置为JobManager pod的用户指定的注释。值可以采用a1:v1,a2:v2的形式。 |
| kubernetes.jobmanager.cpu.amount | 1.0 | Double | Job Manager使用的CPU数量 |
| kubernetes.jobmanager.cpu.limit-factor | 1.0 | Double | Job Manager使用的CPU限制因子。资源限制CPU将设置为cpu * limit-factor。 |
| kubernetes.jobmanager.entrypoint.args | (none) | String | 启动job manager时使用的额外参数。 |
| kubernetes.jobmanager.labels | (none) | Map | 要为JobManager pod设置的标签。以逗号分隔的键值对形式指定。例如，version:alphav1,deploy:test。 |
| kubernetes.jobmanager.memory.limit-factor | 1.0 | Double | Job Manager使用的内存限制因子。资源限制内存将设置为memory * limit-factor。 |
| kubernetes.jobmanager.node-selector | (none) | Map | 要为JobManager pod设置的节点选择器。以逗号分隔的键值对形式指定。例如，environment:production,disk:ssd。 |
| kubernetes.jobmanager.owner.reference | (none) | List | 要设置为JobManager Deployment的用户指定的所有者引用链接。当删除所有所有者资源时，JobManager Deployment将自动被删除，同时也会删除此Flink集群创建的所有资源。该值应格式化为分号分隔的所有者引用列表，其中每个所有者引用是逗号分隔的key:value对。例如，apiVersion:v1,blockOwnerDeletion:true,controller:true,kind:FlinkApplication,name:flink-app-name,uid:flink-app-uid;apiVersion:v1,kind:Deployment,name:deploy-name,uid:deploy-uid |
| kubernetes.jobmanager.replicas | 1 | Integer | 指定同时启动的JobManager pod数量。将该值配置为大于1可以启动备用的JobManagers。这将有助于实现更快的恢复。请注意，在启动备用的JobManagers时应启用高可用性。 |
| kubernetes.jobmanager.service-account | “default” | String | 在Kubernetes集群中由jobmanager使用的服务帐户。当JobManager从API服务器请求taskmanager pod时，job manager会使用此服务帐户。如果未显式配置，将使用配置选项’kubernetes.service-account’。 |
| kubernetes.jobmanager.tolerations | (none) | List | 要设置为JobManager pod的用户指定的宽容度。值应采用以下形式：key:key1,operator:Equal,value:value1,effect:NoSchedule;key:key2,operator:Exists,effect:NoExecute,tolerationSeconds:6000 |
| kubernetes.namespace | “default” | String | 用于运行jobmanager和taskmanager pod的命名空间。 |
| kubernetes.pod-template-file.default | (none) | String | 指定包含pod模板定义的本地文件。它将用于初始化jobmanager和taskmanager pod。主容器应以名称’flink-main-container’定义。请注意，这可以被配置选项’kubernetes.pod-template-file.jobmanager’和’kubernetes.pod-template-file.taskmanager’分别覆盖。 |
| kubernetes.pod-template-file.jobmanager | (none) | String | 指定包含jobmanager pod模板定义的本地文件。它将用于初始化jobmanager pod。主容器应以名称’flink-main-container’定义。如果未显式配置，将使用配置选项’kubernetes.pod-template-file.default’。 |
| kubernetes.pod-template-file.taskmanager | (none) | String | 指定包含taskmanager pod模板定义的本地文件。它将用于初始化taskmanager pod。主容器应以名称’flink-main-container’定义。如果未显式配置，将使用配置选项’kubernetes.pod-template-file.default’。 |
| kubernetes.rest-service.annotations | (none) | Map | 要设置为rest服务的用户指定的注释。值可以采用a1:v1,a2:v2的形式。 |
| kubernetes.rest-service.exposed.node-port-address-type | InternalIP | Enum | 构建node port连接字符串时用于过滤节点IP的地址类型。此选项仅在’kubernetes.rest-service.exposed.type’设置为’NodePort’时考虑。可能的值：“InternalIP”、“ExternalIP” |
| kubernetes.rest-service.exposed.type | ClusterIP | Enum | rest服务的暴露类型。暴露的rest服务可用于访问Flink的Web UI和REST端点。可能的值：“ClusterIP”、“NodePort”、“LoadBalancer”、“Headless_ClusterIP” |
| kubernetes.secrets | (none) | Map | 将挂载到Flink容器中的用户指定的密钥。值应采用foo:/opt/secrets-foo,bar:/opt/secrets-bar的形式。 |
| kubernetes.service-account | “default” | String | 在Kubernetes集群中由jobmanager和taskmanager使用的服务帐户。请注意，这可以通过配置选项’kubernetes.jobmanager.service-account’和’kubernetes.taskmanager.service-account’分别覆盖。 |
| kubernetes.taskmanager.annotations | (none) | Map | 要设置为TaskManager pod的用户指定的注释。值可以采用a1:v1,a2:v2的形式。 |
| kubernetes.taskmanager.cpu.amount | -1.0 | Double | Task Manager使用的CPU数量。默认情况下，将CPU设置为每个TaskManager的slot数。 |
| kubernetes.taskmanager.cpu.limit-factor | 1.0 | Double | Task Manager使用的CPU限制因子。资源限制CPU将设置为cpu * limit-factor。 |
| kubernetes.taskmanager.entrypoint.args | (none) | String | 启动task manager时使用的额外参数。 |
| kubernetes.taskmanager.labels | (none) | Map | 要为TaskManager pods设置的标签。以逗号分隔的键值对形式指定。例如，version:alphav1,deploy:test。 |
| kubernetes.taskmanager.memory.limit-factor | 1.0 | Double | Task Manager使用的内存限制因子。资源限制内存将设置为memory * limit-factor。 |
| kubernetes.taskmanager.node-selector | (none) | Map | 要为TaskManager pods设置的节点选择器。以逗号分隔的键值对形式指定。例如，environment:production,disk:ssd。 |
| kubernetes.taskmanager.service-account | “default” | String | 在Kubernetes集群中由taskmanager使用的服务帐户。当任务管理器从API服务器上的配置映射中获取jobmanager和resourcemanager的leader地址时，任务管理器将使用此服务帐户。如果未显式配置，将使用配置选项’kubernetes.service-account’。 |
| kubernetes.taskmanager.tolerations | (none) | List | 要设置为TaskManager pod的用户指定的宽容度。值应采用以下形式：key:key1,operator:Equal,value:value1,effect:NoSchedule;key:key2,operator:Exists,effect:NoExecute,tolerationSeconds:6000 |
| kubernetes.transactional-operation.max-retries | 5 | Integer | 定义Kubernetes事务操作在客户端放弃之前的重试次数。例如，FlinkKubeClient#checkAndUpdateConfigMap。 |
State Backends
状态后端是Flink用于持久化和管理状态数据的机制。状态后端存储了作业的状态信息，包括中间结果、键值对等。
RocksDB State Backend
RocksDB状态后端是Flink常用的状态后端之一，它基于RocksDB存储引擎实现。以下是一些常用的配置选项：
| Key | Default | Type | Description |
|---|---|---|---|
| state.backend.rocksdb.memory.fixed-per-slot | (none) | MemorySize | 每个slot共享的固定内存总量。当配置了此选项时，它将覆盖’state.backend.rocksdb.memory.managed’选项。 |
| state.backend.rocksdb.memory.fixed-per-tm | (none) | MemorySize | 每个Task Manager共享的固定内存总量（集群级别选项）。仅当未配置’state.backend.rocksdb.memory.managed’和’state.backend.rocksdb.memory.fixed-per-slot’时才生效。如果没有配置，每个RocksDB列族状态都有自己的内存缓存（由列族选项控制）。可在同一级别（flink-conf.yaml）上设置相关资源（例如write-buffer-ratio）。请注意，此功能会破坏slot之间的资源隔离 |
| state.backend.rocksdb.memory.high-prio-pool-ratio | 0.1 | Double | 保留用于高优先级数据（如索引、过滤器和压缩字典块）的缓存内存分数。仅当配置了’state.backend.rocksdb.memory.managed’或’state.backend.rocksdb.memory.fixed-per-slot’时才生效。 |
| state.backend.rocksdb.memory.managed | true | Boolean | 如果设置，RocksDB状态后端将自动配置自身以使用任务槽的托管内存预算，并将内存分配给写缓冲区、索引、块缓存等。这样，RocksDB的三种主要内存使用方式将受到限制。 |
| state.backend.rocksdb.memory.partitioned-index-filters | false | Boolean | 对于分区，SST文件的索引/过滤器块被分成较小的块，并在其上添加了一个额外的顶级索引。在读取索引/过滤器时，只有顶级索引加载到内存中。然后，分区索引/过滤器使用顶级索引根据需要从块缓存加载所需的分区以执行索引/过滤器查询。仅当配置了’state.backend.rocksdb.memory.managed’或’state.backend.rocksdb.memory.fixed-per-slot’时才生效。 |
| state.backend.rocksdb.memory.write-buffer-ratio | 0.5 | Double | 写缓冲区可占用的最大内存量，作为共享内存的一部分。仅当配置了’state.backend.rocksdb.memory.managed’或’state.backend.rocksdb.memory.fixed-per-slot’时才生效。 |
| state.backend.rocksdb.timer-service.factory | ROCKSDB | Enum | 确定计时器服务状态实现的工厂。可能的值：“HEAP”：基于堆的；“ROCKSDB”：基于RocksDB |
Metrics
请参考Flink指标系统文档了解Flink的指标基础设施。
| Key | Default | Type | Description |
|---|---|---|---|
| metrics.fetcher.update-interval | 10000 | Long | 指标获取器（用于Web UI）的更新间隔，以毫秒为单位。如果希望更快地更新指标，请减小该值。如果指标获取器造成过大负载，请增大该值。将此值设置为0会完全禁用指标获取。 |
| metrics.internal.query-service.port | “0” | String | Flink内部指标查询服务使用的端口范围。可接受端口列表（“50100,50101”）、范围（“50100-50200”）或两者结合。建议设置一段端口范围，以避免多个Flink组件在同一台机器上运行时发生冲突。默认情况下，Flink将选择一个随机端口。 |
| metrics.internal.query-service.thread-priority | 1 | Integer | Flink内部指标查询服务使用的线程优先级。该线程由Akka的线程池执行器创建。优先级的范围是从1（MIN_PRIORITY）到10（MAX_PRIORITY）。警告：增加此值可能导致主要的Flink组件停止工作。 |
| metrics.job.status.enable | CURRENT_TIME | List | 应报告的作业状态指标的选择。可能的值：“STATE”：对于给定的状态，如果作业当前处于该状态，则返回1；否则返回0。“CURRENT_TIME”：对于给定的状态，如果作业当前处于该状态，则返回自从作业转换为该状态以来的时间；否则返回0。“TOTAL_TIME”：对于给定的状态，返回作业在该状态下总共花费的时间。 |
| metrics.latency.granularity | “operator” | String | 定义延迟指标的粒度。可接受的值为：single - 跟踪延迟而不区分来源和子任务。operator - 跟踪延迟并区分来源，但不区分子任务。subtask - 跟踪延迟并区分来源和子任务。 |
| metrics.latency.history-size | 128 | Integer | 定义每个算子保留的测量延迟的数量。 |
| metrics.latency.interval | 0 | Long | 定义源发出的延迟跟踪标记的间隔。如果设置为0或负值，则禁用延迟跟踪。启用此功能可能会显著影响集群的性能。 |
| metrics.reporter.. | (none) | String | 为名为的报告器配置参数。 |
| metrics.reporter..factory.class | (none) | String | 用于名为的报告器的报告器工厂类。 |
| metrics.reporter..filter.excludes | List | 名为的报告器应该排除的指标。格式与filter.includes相同。 |
|
| metrics.reporter..filter.includes | “:😗” | List | 名为的报告器应该包含的指标。过滤器指定为列表，每个过滤器遵循以下格式：<scope>[:<name>[,<name>][:<type>[,<type>]]]如果范围模式和至少一个名称模式以及至少一个类型匹配，则指标与过滤器匹配。范围：基于逻辑范围进行过滤。以匹配任何字符序列的“”为模式，“.”分隔范围组件。例如：“jobmanager.job”匹配JobManager上的任何与作业相关的指标，“*.job”匹配所有与作业相关的指标，“*.job.*”匹配作业级别以下的所有指标（即任务/算子指标等）。名称：基于指标名称进行过滤。以逗号分隔的模式列表，其中“”匹配任何字符序列。例如，“*Records*”匹配名称包含“Records”或“Bytes”的任何指标。类型：基于指标类型进行过滤。以逗号分隔的指标类型列表：[counter, meter, gauge, histogram]示例：“*:numRecords*”匹配numRecordsIn之类的指标。“*.job.task.operator:numRecords*”匹配运算符级别上的numRecordsIn之类的指标。“*.job.task.operator:numRecords*:meter”匹配运算符级别上的numRecordsInPerSecond之类的计量器指标。“*:numRecords*,numBytes*:counter,meter”匹配所有计数器/计量器指标，例如numRecordsInPerSecond。 |
| metrics.reporter..interval | 10 s | Duration | 名为的报告器使用的报告间隔。仅适用于基于推送的报告器。 |
| metrics.reporter..scope.delimiter | “.” | String | 用于组装名为的报告器的指标标识符的分隔符。 |
| metrics.reporter..scope.variables.additional | Map | 应包含在名为的报告器中的其他变量的映射。仅适用于基于标签的报告器。 | |
| metrics.reporter..scope.variables.excludes | “.” | String | 对于名为的报告器应排除的一组变量。仅适用于基于标签的报告器。 |
| metrics.reporters | (none) | String | 可选的报告器名称列表。如果配置了此项，则仅启动与列表中任何名称匹配的报告器。否则，将启动在配置中找到的所有报告器。 |
| metrics.scope.delimiter | “.” | String | 用于组装指标标识符的分隔符。 |
| metrics.scope.jm | “.jobmanager” | String | 应用于所有作用域为JobManager的指标的范围格式字符串。仅在配置了基于标识符的报告器时有效。 |
| metrics.scope.jm-job | “.jobmanager.<job_name>” | String | 应用于JobManager上作业的所有指标的范围格式字符串。仅在配置了基于标识符的报告器时有效。 |
| metrics.scope.operator | “.taskmanager.<tm_id>.<job_name>.<operator_name>.<subtask_index>” | String | 应用于算子的所有指标的范围格式字符串。仅在配置了基于标识符的报告器时有效。 |
| metrics.scope.task | “.taskmanager.<tm_id>.<job_name>.<task_name>.<subtask_index>” | String | 应用于任务的所有指标的范围格式字符串。仅在配置了基于标识符的报告器时有效。 |
| metrics.scope.tm | “.taskmanager.<tm_id>” | String | 应用于所有作用域为TaskManager的指标的范围格式字符串。仅在配置了基于标识符的报告器时有效。 |
| metrics.scope.tm-job | “.taskmanager.<tm_id>.<job_name>” | String | 应用于TaskManager上作业的所有指标的范围格式字符串。仅在配置了基于标识符的报告器时有效。 |
| metrics.system-resource | false | Boolean | 标志，指示Flink是否报告系统资源指标，例如机器的CPU、内存或网络使用情况。 |
| metrics.system-resource-probing-interval | 5000 | Long | 指定以毫秒为单位的系统资源指标探测间隔。仅当启用’metrics.system-resource’时才生效。 |
RocksDB本地指标
对于使用RocksDB状态后端的应用程序，Flink可以报告RocksDB的本地代码中的指标。这些指标针对无符号长整型的运算符进行作用，并具有两种类型：
- 基于RocksDB属性的指标，按列族细分，例如某个特定列族当前正在运行的压缩数。
- 基于RocksDB统计的指标，在数据库级别保持，例如DB内的总块缓存命中次数。
启用RocksDB的本地指标可能会导致性能下降，请谨慎设置。
| Key | Default | Type | Description |
|---|---|---|---|
| state.backend.rocksdb.metrics.actual-delayed-write-rate | false | Boolean | 监视当前实际延迟写入速率。0表示无延迟。 |
| state.backend.rocksdb.metrics.background-errors | false | Boolean | 监视RocksDB中的后台错误数。 |
| state.backend.rocksdb.metrics.block-cache-capacity | false | Boolean | 监视块缓存容量。 |
| state.backend.rocksdb.metrics.block-cache-hit | false | Boolean | 监视RocksDB中块缓存的总命中次数（BLOCK_CACHE_HIT == BLOCK_CACHE_INDEX_HIT + BLOCK_CACHE_FILTER_HIT + BLOCK_CACHE_DATA_HIT）。 |
| state.backend.rocksdb.metrics.block-cache-miss | false | Boolean | 监视RocksDB中块缓存的总未命中次数（BLOCK_CACHE_MISS == BLOCK_CACHE_INDEX_MISS + BLOCK_CACHE_FILTER_MISS + BLOCK_CACHE_DATA_MISS）。 |
| state.backend.rocksdb.metrics.block-cache-pinned-usage | false | Boolean | 监视块缓存中被固定的条目的内存大小。 |
| state.backend.rocksdb.metrics.block-cache-usage | false | Boolean | 监视块缓存中驻留的条目的内存大小。 |
| state.backend.rocksdb.metrics.bytes-read | false | Boolean | 监视RocksDB的Get()操作从内存表/缓存/SST中读取的未压缩字节数。 |
| state.backend.rocksdb.metrics.bytes-written | false | Boolean | 监视RocksDB的DB::{Put(), Delete(), Merge(), Write()}操作写入的未压缩字节数，不包括压缩写入字节。 |
| state.backend.rocksdb.metrics.column-family-as-variable | false | Boolean | 是否将列族作为变量公开给基于属性的RocksDB指标。 |
| state.backend.rocksdb.metrics.compaction-pending | false | Boolean | 跟踪RocksDB中待处理的压缩数。如果存在待处理的压缩，则返回1，否则返回0。 |
| state.backend.rocksdb.metrics.compaction-read-bytes | false | Boolean | 监视RocksDB中压缩期间读取的字节数。 |
| state.backend.rocksdb.metrics.compaction-write-bytes | false | Boolean | 监视RocksDB中压缩期间写入的字节数。 |
| state.backend.rocksdb.metrics.cur-size-active-mem-table | false | Boolean | 监视活跃内存表的近似大小（以字节为单位）。 |
| state.backend.rocksdb.metrics.cur-size-all-mem-tables | false | Boolean | 监视活跃且未刷新的不可变内存表的近似总大小（以字节为单位）。 |
| state.backend.rocksdb.metrics.estimate-live-data-size | false | Boolean | 估计活动数据的大小（以字节为单位）（由于空间扩大，通常小于sst文件的大小）。 |
| state.backend.rocksdb.metrics.estimate-num-keys | false | Boolean | 估计RocksDB中键的数量。 |
| state.backend.rocksdb.metrics.estimate-pending-compaction-bytes | false | Boolean | 估计压缩需要重写的字节数，以使所有级别的大小均低于目标大小。不适用于除基于级别的压缩之外的其他压缩。 |
| state.backend.rocksdb.metrics.estimate-table-readers-mem | false | Boolean | 估计用于读取SST表的内存使用量（不包括块缓存中使用的内存，例如过滤器和索引块）（以字节为单位）。 |
| state.backend.rocksdb.metrics.is-write-stopped | false | Boolean | 跟踪RocksDB中是否已停止写入。如果已停止写入，则返回1，否则返回0。 |
| state.backend.rocksdb.metrics.iter-bytes-read | false | Boolean | 监视从迭代器操作中RocksDB中未压缩读取的字节数（从内存表/缓存/SST中读取）。 |
| state.backend.rocksdb.metrics.live-sst-files-size | false | Boolean | 监视最新版本的所有SST文件的总大小（以字节为单位）。警告：如果文件太多，可能会减慢在线查询。 |
| state.backend.rocksdb.metrics.mem-table-flush-pending | false | Boolean | 监视RocksDB中挂起的内存表刷新数。 |
| state.backend.rocksdb.metrics.num-deletes-active-mem-table | false | Boolean | 监视活动内存表中的删除条目总数。 |
| state.backend.rocksdb.metrics.num-deletes-imm-mem-tables | false | Boolean | 监视未刷新的不可变内存表中的删除条目总数。 |
| state.backend.rocksdb.metrics.num-entries-active-mem-table | false | Boolean | 监视活跃内存表中的条目总数。 |
| state.backend.rocksdb.metrics.num-entries-imm-mem-tables | false | Boolean | 监视未刷新的不可变内存表中的条目总数。 |
| state.backend.rocksdb.metrics.num-immutable-mem-table | false | Boolean | 监视RocksDB中的不可变内存表数。 |
| state.backend.rocksdb.metrics.num-live-versions | false | Boolean | 监视活动版本的数量。版本是一个内部数据结构。有关详细信息，请参见RocksDB文件version_set.h。较多的活动版本通常意味着更多的SST文件被迭代器或未完成的压缩所占用，无法被删除。 |
| state.backend.rocksdb.metrics.num-running-compactions | false | Boolean | 监视当前正在运行的压缩数。 |
| state.backend.rocksdb.metrics.num-running-flushes | false | Boolean | 监视当前正在运行的刷新数。 |
| state.backend.rocksdb.metrics.num-snapshots | false | Boolean | 监视数据库的未释放快照数。 |
| state.backend.rocksdb.metrics.size-all-mem-tables | false | Boolean | 监视活跃、未刷新的不可变内存表和固定不可变内存表的近似总大小（以字节为单位）。 |
| state.backend.rocksdb.metrics.stall-micros | false | Boolean | 监视RocksDB中等待压缩或刷新结束的写入所需的持续时间。 |
| state.backend.rocksdb.metrics.total-sst-files-size | false | Boolean | 监视所有版本的所有SST文件的总大小（以字节为单位）。警告：如果文件太多，可能会减慢在线查询。 |
历史服务器
历史服务器保存已完成作业的信息（图形、运行时、统计数据）。要启用它，您需要在JobManager中启用“作业归档”(jobmanager.archive.fs.dir)。文章来源：https://www.toymoban.com/news/detail-848531.html
有关详细信息，请参见历史服务器文档。文章来源地址https://www.toymoban.com/news/detail-848531.html
| Key | Default | Type | Description |
|---|---|---|---|
| historyserver.archive.clean-expired-jobs | false | Boolean | HistoryServer是否应清理不再存在于historyserver.archive.fs.dir中的作业。 |
| historyserver.archive.fs.dir | (none) | String | 用于获取归档作业的目录列表。历史服务器将监视这些目录以获取归档作业。您可以通过jobmanager.archive.fs.dir配置JobManager将作业归档到一个目录。 |
| historyserver.archive.fs.refresh-interval | 10000 | Long | 刷新归档作业目录的间隔（以毫秒为单位）。 |
| historyserver.archive.retained-jobs | -1 | Integer | 每个由historyserver.archive.fs.dir定义的归档目录中要保留的作业的最大数量。如果设置为-1（默认值），则没有限制作为归档数量。如果设置为0或小于-1，HistoryServer将抛出IllegalConfigurationException。 |
| historyserver.log.jobmanager.url-pattern | (none) | String | JobManager日志URL的模式。历史服务器将从中生成实际URL，并将特殊占位符“”替换为作业ID。仅支持http / https方案。 |
| historyserver.log.taskmanager.url-pattern | (none) | String | TaskManager日志URL的模式。历史服务器将从中生成实际URL，并将特殊占位符“”和“”分别替换为作业和TaskManager的ID。仅支持http / https方案。 |
| historyserver.web.address | (none) | String | 历史服务器Web界面的地址。 |
| historyserver.web.port | 8082 | Integer | 历史服务器Web界面的端口。 |
| historyserver.web.refresh-interval | 10000 | Long | 历史服务器Web前端刷新间隔（以毫秒为单位）。 |
| historyserver.web.ssl.enabled | false | Boolean | 启用对历史服务器Web前端的HTTPs访问。仅在全局SSL标志security.ssl.enabled设置为true时有效。 |
| historyserver.web.tmpdir | (none) | String | 由历史服务器REST API用于临时文件的本地目录。 |
到了这里，关于【flink配置系列】FLink配置大全的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
