当前位置：首页 > article >正文

flink 内存配置（五）：网络缓存调优

article 2024/11/12 13:25:43

flink 内存配置（一）：设置Flink进程内存

flink 内存配置（二）：设置TaskManager内存

flink 内存配置（三）：设置JobManager内存

flink 内存配置（四）：内存调优和问题处理

flink 内存配置（五）：网络缓存调优

1. Network Buffer总览

Flink中的每条记录都会与网络缓冲区（network buffer）中的其他记录一起发送给下一个子任务，网络缓冲区是子任务之间通信的最小单位。为了保持一致的高吞吐量，Flink在传输过程的输入和输出侧使用网络缓冲队列（也被称为传输中数据 in-flight data）。

每个子任务都有一个等待消费数据的输入队列和一个等待将数据发送到下一个子任务的输出队列。拥有更多的传输中数据（in-flight data）意味着Flink可以在管道中提供更高、更有弹性的吞吐量。然而，这将导致检查点时间更长。

Flink中的检查点只能在所有子任务接收到所有注入的检查点屏障（checkpoint barriers）后才能完成。在对齐的检查点中，这些检查点屏障与网络缓冲区一起在作业图中移动。传输中的数据量越大，检查点屏障传播时间越长。在未对齐的检查点中，传输中的数据量越大，检查点的大小就越大，因为所有捕获的传输中数据都必须作为检查点的一部分进行持久化。

2. 缓冲区去膨胀（debloat）机制

在以前，配置传输中数据量的唯一方法是指定缓冲区数量和缓冲区大小。然而，理想值可能很难选择，因为它们对于每个部署应用都是不同的。Flink 1.14中添加的缓冲区去膨胀机制试图通过自动将传输中的数据量调整到合理的值来解决这个问题。

缓冲区去膨胀功能计算子任务的最大可能吞吐量（在它总是繁忙的情况下），并调整正在传输的数据量，使这些正在传输数据的消耗时间等于配置值。

通过将属性taskmanager.network.memory.buffer-debloat.enabled设置为true，可以启用缓冲区去膨胀机制。可以通过属性 taskmanager.network.memory.buffer-debloat.target配置传输中数据的目标消耗时间。buffer-debloat.target配置的默认值应该要适应大多数的情况。

此功能使用过去的吞吐量数据来预测消耗剩余传输中数据所需的时间。如果预测不正确，去膨胀机制可能会以以下两种方式之一失败：

没有足够的缓冲数据来提供完整的吞吐量。
缓存的传输中数据过多，会对对齐检查点屏障的传播时间或非对齐检查点大小产生负面影响。

如果你有一个负载会随着时间变化的作业（比如输入的记录突然激增，周期性地触发窗口聚合或连接），你可能就需要调整以下设置：

taskmanager.network.memory.buffer-debloat.period - 这是重新计算缓冲区大小的最小时间间隔。周期越短，去膨胀机制的反应时间越快，但所需计算的CPU开销越高。
taskmanager.network.memory.buffer-debloat.samples - 调整吞吐量测量的平均样本数量。收集样本的频率可以通过taskmanager.network.memory.buffer-debloat.period进行调整。样本越少，去膨胀机制的反应时间越快，但吞吐量突然上升或下降的可能性越大，这可能导致缓冲区去膨胀机制错误计算最佳传输数据量。
taskmanager.network.memory.buffer-debloat.threshold-percentages - 防止频繁改变缓冲区大小的优化（例如，如果新的大小与旧的大小没有太大的不同）。

以下是可用于监控当前缓冲区大小的指标：

estimatedTimeToConsumeBuffersMs - 消耗所有输入通道数据的总时间
debloatedBufferSize - 当前缓冲区大小

3. 限制

目前，有一些情况不是由缓冲区去膨胀机制自动处理的。

Multiple inputs and unions

目前，吞吐量计算和缓冲区去膨胀都发生在子任务级别。

如果你的子任务有多个不同的输入，或者只有一个但是是union的输入，那么缓冲区去膨胀可能会导致低吞吐量的输入有太多缓冲的传输中数据，而高吞吐量的输入可能有太小的缓冲区，无法维持该吞吐量。如果不同的输入具有不同的吞吐量，这可能特别明显。我们建议在测试此功能时特别注意此类子任务。

Buffer size and number of buffers

当前，缓冲区去膨胀只会限制使用的最大缓冲区大小。实际的缓冲区大小和缓冲区数量保持不变。这意味着去膨胀机制不能减少作业的内存使用。你必须手动减少缓冲区的数量或大小。

此外，如果你希望将缓冲的传输数据量减少到缓冲区去膨胀当前允许的范围以下，则可能需要手动配置缓冲区的数量。

High parallelism

目前，使用默认配置，缓冲区去膨胀机制可能无法在高并行性（高于~200）下正确执行。如果你观察到吞吐量降低或检查点时间高于预期，我们建议将浮动缓冲区的数量（taskmanager.network.memory.floating-buffers-per-gate）从默认值增加到至少等于并行度的数量。

该问题发生的并行性的实际值因作业而异，但通常应该超过几百。

4. Network Buffer生命周期

Flink有几个本地缓冲池（buffer pools）——一个用于输出流，一个用于每个输入门。每个缓冲池的目标大小由以下公式计算：

channels * taskmanager.network.memory.buffers-per-channel + taskmanager.network.memory.floating-buffers-per-gate

缓冲区的大小可以通过设置taskmanager.memory.segment-size来配置。

输入网络缓冲区

目标缓冲池的大小并不总是达到。有一个阈值控制着Flink在无法获得缓冲区时是否会失败。在缓冲区的目标数目中，低于该阈值的部分被认为是必需的。剩下的（如果有的话）是可选的。没有获得所需的缓冲区将导致任务失败。如果无法获得可选的缓冲区，任务不会失败，但可能会降低性能。

这个阈值的默认值是Integer.MAX_VALUE用于流工作负载，1000用于批处理工作负载。我们不建议用户改变这个阈值，除非用户有很好的理由并且知道他/她做得很好。相关的配置选项是taskmanager.network.memory.read-buffer.required-per-gate.max。一般来说，阈值越小，出现“insufficient number of network buffers”异常的可能性越小，而工作负载可能会悄无声息地受到性能影响，反之亦然。

输出网络缓冲区

与输入缓冲池不同，输出缓冲池只有一种类型的缓冲区，它在所有子分区之间共享。

为了避免过度的数据倾斜，每个子分区的缓冲区数量受到taskmanager.network.memory.max-buffers-per-channel设置的限制。

与输入缓冲池不同，配置的独占缓冲区和浮动缓冲区的数量只被视为推荐值。如果没有足够的缓冲区可用，Flink可以在每个输出子分区只有一个独占缓冲区、没有浮动缓冲区的情况下继续工作。

透支缓冲区

对于每个输出子任务，也可以请求taskmanager.network.memory.max-overdraft-buffers-per-gate（默认为5）额外的透支缓冲区。只有当子任务受到下游子任务的反向压力，并且子任务需要多个网络缓冲区来完成其当前工作时，才会使用这些缓冲区。这种情况可能发生在：

序列化一个网络缓冲区容纳不了的非常大的记录。
类似于Flat Map操作符，每个输入记录产生许多输出记录。
周期性地或根据某些事件（例如WindowOperator的触发器）输出许多记录的操作符。

在这种情况下，如果缓冲区没有透支，Flink的子任务线程就会因为负载压力而阻塞，从而导致未对齐的检查点无法完成。为了缓解这个问题，我们添加了透支缓冲区的概念。这些透支缓冲区是严格可选的，Flink可以逐步只使用普通缓冲区进行处理，这意味着对于taskmanager.network.memory.max-overdraft-buffers-per-gate来说，0是一个可以接受的配置。注意透支缓冲区只对 Pipelined Shuffle有用。