新的Task Executor内存模型

FLIP49为Task Executor引入了一种新的内存模型。引入了新的配置选项来控制Task Executor的内存消耗。将影响所有类型的部署:standalone, YARN, Mesos和Kubernetes。job manager的内存模型尚未更改，但也计划进行更新。

如果尝试在不做任何调整的情况下重用以前的Flink配置，新的内存模型可能会导致JVM的计算内存参数不同，从而导致性能变化。

FLIP-49 Task Executor统一内存配置

该方案解决了Flink 1.9 TaskExecutor内存配置的几个缺点。

1. 流和批配置不同

目前，TaskExecutor内存的配置对于流作业和批作业是不同的。

• Streaming
  
• 内存是隐式消耗的，要么在堆内由Memory State Backend消耗，要么在堆外由RocksDB消耗。
• 用户必须手动调整堆大小和状态后端选择。
• 用户必须手动配置RocksDB，以使用足够的内存实现良好的性能，但又不会超过预算。
• 内存消耗不可预测，包括堆内的Memory State Backend，和堆外的RocksDB


• Batch
  
• 用户配置总内存大小，以及在算子中使用堆内存还是堆外内存。
  
• Flink将总内存的一部分保留为托管内存(managed memory).它自动调整堆大小和“max direct memory”参数，以适应托管内存在堆内或堆外的情况。
  
• Flink为算子使用的托管内存分配内存段。它保证不会超过剩余的内存段。
  

2.流计算中RocksDB的配置复杂困难

• 用户必须手动减少JVM堆大小，或者将Flink设置为使用非堆内存。
• 用户必须手动配置RocksDB内存。
• 用户没有办法尽可能多地使用可用内存，因为RocksDB内存大小必须配置得足够低，以确保不会超出内存预算

3.复杂、不确定、难以理解

• 在决定containers、进程的内存大小时，有一些“magic”。其中一些是不容易理解的，例如“cutoff ratio”(Container预留的非TM内存占设定的TM内存的比例，默认值0.25)在YARN或其他containers。
  
• 配置一个像RocksDB这样的堆外状态后端意味着要么将托管内存设置为堆外，要么调整cutoff ratio，从而将更少的内存分配给JVM堆。
  
• TaskExecutor依赖于瞬时JVM内存使用来确定不同内存池的大小，首先触发GC，然后获得JVM空闲内存大小，这给不同内存池的大小带来了不确定性。
  

Flink1.10内存模型

RocksDB状态后端内存控制

Flink的RocksDB状态后端分配native内存。RocksDB分配的内存量不受Flink或JVM的控制，(理论上)可以无限制地增长。在容器环境中，这可能会有问题，因为进程可能超出容器的内存预算，并且进程将被杀死。除了相信RocksDB表现良好并遵循其内存配置之外，没有其他选择。然而，限制RocksDB的内存使用并不像设置limit参数那么容易。内存限制是由几个配置参数的相互作用决定的，这对于用户来说几乎是不可能的。更糟糕的是，多个RocksDB实例可以在同一个进程中运行，并根据Flink作业对配置进行推理。

随着新Task Executor内存模型的引入，RocksDB状态后端的内存消耗将受到Flink托管内存总量的限制。可通过 taskmanager.memory.managed.size或taskmanager.memory.managed.fraction配置。此外，用户可以调优RocksDB的写/读内存比率(state.backend.rocksdb.memory.write-buffer-ratio，默认为0.5)以及为索引/过滤器保留的内存部分 (state.backend.rocksdb.memory.high-prio-pool-ratio, 默认 0.1).

细粒度算子资源管理

配置项 table.exec.resource.external-buffer-memory, table.exec.resource.hash-agg.memory, table.exec.resource.hash-join.memory, table.exec.resource.sort.memory 已经弃用。从Flink 1.10开始，这些配置选项被解释为weight hints，而不是绝对内存需求。Flink选择了合理的默认weight hints，用户不应该对其进行调整。

这是“FLIP-53:细粒度算子资源管理”讨论的核心问题。