Flink系列知识讲解之：网络监控、指标与反压

Flink系列知识之：网络监控、指标与反压

在上一篇博文中，我们介绍了 Flink 网络协议栈从高层抽象到底层细节的工作原理。本篇博文是网络协议栈系列博文中的第二篇，在此基础上，我们将讨论如何监控网络相关指标，以识别吞吐量和延迟方面的反压或瓶颈等影响。如果您对网络协议栈不熟悉，我们强烈建议您先阅读网络协议栈深度剖析，然后继续阅读本文。

网络监控

网络监控最重要的部分可能是监控反压，即操作符算子接收数据的速度超过其处理速度的情况¹。这种行为会导致上游发送方受到反压，原因可能有两个：

• 接收器处理性能较慢
  • 出现这种情况的原因可能是接收器本身受到了反压，无法保持与发送器相同的处理速度，或者由于垃圾回收、系统资源不足或 I/O 而暂时受阻。
• 网络通道速度较慢
  • 尽管在这种情况下接收方并没有（直接）参与，但由于在同一台机器上运行的所有子任务共享的网络带宽可能被超量占用，因此我们称发送者受到了反向压力。请注意，除了 Flink 的网络堆栈，可能还有更多的网络用户，如source和sink，分布式文件系统（检查点、网络附加存储）、日志和指标数据都会在网络通道中传输。

如果出现反压，这种压力会向上游传递，最终到达Flink应用的源头，从而使其变慢。这种反压机制本身并不是坏事，某种程度来说是应用程序根据当前资源情况的一种自我保护机制，只是说明您缺乏资源来应对当前的负载。不过，您可能会希望通过改进应用程序，使其能够在不使用更多资源的情况下应对更高的负载。
为此，您需要找到：
(1) 瓶颈在哪里（哪个任务/操作符算子）；
(2) 造成瓶颈的原因。Flink 提供两种机制来识别瓶颈所在：

• 直接通过 Flink 的网络用户界面（Flink web UI）和反压监控器（backpressure monitor）
• 间接地通过一些网络指标。

Flink Web UI可能是快速排除故障的第一入口，但也有一些缺点，我们将在下文中解释。另一方面，Flink 的网络指标（netwrok metrics）更适合持续监控和推理造成反压的瓶颈的确切问题所在。我们将在下面的章节中介绍这两种方法。在这两种情况下，您都需要确定从source到sink的背压具体来源。值得注意的是，反压的根源节点并不一定会在反压面板体现出高反压，因为反压面板监控的是发送端，如果某个节点是性能瓶颈并不会导致它本身出现高反压，而是导致它的上游出现高反压。总体来看，如果我们找到第一个出现反压的节点，那么反压根源要么是就这个节点，要么是它紧接着的下游节点。

反压监控

反压监控可以通过 Flink 的 Web UI² 进行监控。原理是反压监控器周期性地通过 Thread.getStackTrace() 对所有TaskManager上的运行Task线程的栈信息进行采样，得到线程被阻塞在请求 Buffer（意味着被下游队列阻塞）的频率来判断该节点是否处于反压状态。这些Task要么是无法以buffer的生产速度被发送到网络缓冲区，要么是下游Task处理缓冲区的速度很慢，导致基于credit-based流量控制时，返回给上游发送端的credit值较小（或等于0）。
反压监控器（backpressure monitor）将显示被阻止的请求buffer与总请求的比率。由于某些反压被认为是正常/临时的，因此它将显示以下状态：

• OK for ratio ≤ 0.10,
• LOW for 0.10 < Ratio ≤ 0.5, and
• HIGH for 0.5 < Ratio ≤ 1.

虽然可以调整刷新间隔、采样次数或采样延迟等参数，但通常情况下，您并不需要调整这些参数，因为默认值已经能提供足够好的结果。

也可以通过 Flink提供的REST API 接口来访问反压监控指标：/jobs/:jobid/vertices/:vertexid/backpressure

反压监控器可以帮助您找到反压的来源（在哪个任务/操作符算子处）。但是，它无法帮助您进一步推理产生反压的原因。此外，对于较大的任务或并行度较高的任务，反压监控器会变得过于拥挤而无法使用，而且从所有TaskManager收集所有信息也可能需要一些时间。另外需要注意，频繁地对Task线程栈采样也可能会影响运行任务的性能。

网络指标—Network Metrics

网络（Network）和任务 I/O （Task I/O）指标比反压监控更轻量级，而且会持续地为每个正在运行的作业进行监控。我们可以利用这些指标，获得更多信息，而不仅仅是用于反压监控。与用户最相关的指标包括：

• 至 Flink 1.8：outPoolUsage、inPoolUsage
  • 评估已使用的缓冲区与各本地缓冲区中可用缓冲区的比率。在使用基于credit-based的流量控制的 Flink 1.5 - 1.8 版本中，inPoolUsage只与浮动缓冲区有关（独占缓冲区不是池的一部分）。
• Flink 1.9以及更高：outPoolUsage, inPoolUsage, floatingBuffersUsage, exclusiveBuffersUsage
  • 评估已使用的缓冲区与各本地缓冲区中可用缓冲区的比率。从 Flink 1.9 开始，inPoolUsage 是 floatingBuffersUsage 和 exclusiveBuffersUsage 的总和。
• numRecordsOut, numRecordsIn
  • 该指标包含两个作用域：一个作用域针对操作符算子，另一个作用域针对subtask子任务。对于网络监控来说，子任务域显示的是SubTask发送/接收的记录总数。在实际使用时，需要指定具体的时间跨度…PerSecond，比如numRecordsOutPerSecond、numRecordsInPerSecond。
• numBytesOut, numBytesInLocal, numBytesInRemote
  • 该SubTask从本地/远程发出或读取的字节总数。这些数据也可以通过 …PerSecond 指标作为时间跨度提供。
• numBuffersOut, numBuffersInLocal, numBuffersInRemote
  • 与 numBytes... 类似，但计算的是网络缓冲区的数量。

另外需要注意：

• 任何至少被使用过一次的输出通道都会占用一个缓冲区（自 Flink 1.5 起）。
  • 直到 Flink 1.8：该缓冲区（即使是空的）也会始终被算作值为1的积压，因此接收器试图为其保留一个浮动缓冲区。
  • Flink 1.9 及以上版本：缓冲区只有在可以使用（即已满或已刷新）时才会被计入积压（参见FLINK-11082）
• 接收器只有在反序列化缓冲区中的最后一条记录后，才会释放接收到的缓冲区。

以下各节将利用并结合这些指标来推理与吞吐量有关的反压和资源使用/效率。另一部分将详细介绍与延迟相关的指标。

Backpressure

反压（backpressure）是实时计算应用开发中，特别是流式计算中，十分常见的问题。反压意味着数据管道中某个节点成为瓶颈，处理速率跟不上上游发送数据的速率，而需要对上游进行限速。由于实时计算应用通常使用消息队列来进行生产端和消费端的解耦，消费端数据源是 pull-based 的，所以反压通常是从某个节点传导至数据源并降低数据源（比如 Kafka consumer）的摄入速率。

我们将在下文中重点讨论输入和输出池的使用情况对反压的监控：

• 如果SubTask的 outPoolUsage 为 100%，它本身就已经受到反向压力。子任务是已经阻塞还是仍在向网络缓冲区写入记录，取决于RecordWriters当前写入的缓冲区有多满。
• inPoolUsage 为 100% 意味着所有浮动缓冲区都已经分配给了输入通道，最终将对上游传导反压。这些浮动缓冲区可能处于以下任一情况：
  • 由于独占缓冲区被使用，它们被保留以此在未来被某一通道使用
  • 它们被保留用来接收发送方的积压数据（backlog）并等待数据
  • 它们可能包含数据并在输入通道中排队
  • 或者它们可能包含数据并正在被接收方的子任务读取（一次读取一条记录）。
• 至 Flink 1.8：由于 FLINK-11082，即使在正常情况下，inPoolUsage 达到 100% 也很常见。
• Flink 1.9以及更高：如果 inPoolUsage 持续保持在 100% 左右，则表明上游正在经受反压。

下表总结了inPoolUsage和outPoolUsage的所有组合及其解释。但请记住，反压可能是轻微的或暂时的（无需调查），只存在于特定通道上，或由于特定TaskManager的其他 JVM 进程造成的，如 GC、同步、I/O、资源短缺，而非特定的子任务。

我们甚至可以通过观察两个连续任务的子任务的网络指标来进一步推断反压产生的原因：

• 如果接收方任务的所有子任务的 inPoolUsage 值都很低，而上游任务子任务的 outPoolUsage 值都很高，那么可能是网络瓶颈导致了反压。由于网络是TaskManager所有子任务的共享资源，因此这种反压的产生可能不是直接来自该子任务，而是来自其他各种并发操作，如checkpoint、其他流、外部连接或同一台机器上的其他TaskManager/进程。

反压也可能由任务的所有并行实例或单个任务实例造成。前者通常是因为任务在执行某些针对所有输入分区都耗时的操作。后者通常是某种倾斜的结果，可能是shuffle过程中的数据倾斜，也可能是资源可用性/分配的倾斜。

Flink 1.9以及更高

• 如果 floatingBuffersUsage 不是 100%，则不太可能存在反压。如果浮动缓冲区使用率为 100%，且任何上游任务都存在反压，则表明该输入通道正在对单个、部分或所有输入通道施加反压。要区分这三种情况，可以使用 exclusiveBuffersUsage。
• 假设 floatingBuffersUsage 约为 100%，则 exclusiveBuffersUsage 越高，正在经历反压的输入通道就越多（exclusiveBuffersUsage每个输入通道独占）。在 exclusiveBuffersUsage 接近 100% 的极端情况下，这意味着当前所有输入通道都在受到反压。

下表概括了 “exclusiveBuffersUsage”、"floatingBuffersUsage "和上游任务的 "outPoolUsage "之间的关系，并在上表的基础上扩展了 “inPoolUsage = floatingBuffersUsage + exclusiveBuffersUsage”：

资源使用量 / 吞吐量—Resource Usage / Throughput

除了上述每个监控指标的明显用途外，还有一些组合指标可以帮助我们深入了解网络堆栈中发生的情况：

• 低吞吐量，outPoolUsage 值经常在 100%左右，但所有接收器的 inPoolUsage 值都很低，这说明在做流量控制时的credit-notification（取决于网络延迟）的往返时间过长，不足以利用带宽。可以考虑增加taskmanager.network.memory.buffers-per-channel参数，或尝试禁用基于credit-based的流量控制（不推荐）。
• 将 numRecordsOut 和 numBytesOut 结合起来，有助于确定平均序列化记录大小，从而可以针对峰值情况进行容量规划。
• 如果要推理缓冲区填充率（buffer fill rate）和输出刷新器（output flusher）的影响，可以将 numBytesInRemote 与 numBuffersInRemote 结合起来。
  • 当希望调整吞吐量（而不是延迟）时，低缓冲区填充率可能表明网络效率降低。在这种情况下，可以考虑增加缓冲区超时时间，以此来增加单次发送的数据量来提升吞吐。请注意，从 Flink 1.8 和 1.9 版开始，numBuffersOut 只有在缓冲区满或特殊事件（如checkpoint barrier）时才会触发增加，而且可能会滞后。另外需要注意，对本地通道的缓冲区填充率进行推理是不必要的，因为缓冲是一种针对远程通道的优化技术，对本地通道的影响有限。
• 也可以使用 numBytesInLocal 和 numBytesInRemote 将本地流量和远程流量分开，但在大多数情况下这是不必要的。

反压产生的潜在原因

假设已经确定了反压产生的源头（瓶颈）所在，下一步就是分析发生反压的原因。下面，我们列出了一些可能造成反压的潜在原因，从较基本的到较复杂的。建议首先检查基本原因，然后再深入研究更复杂的原因，以免得出错误的结论。

同时需要注意，反压可能是暂时的，是负载峰值、触发checkpoint或作业重启时有积压数据等待处理的结果。在这种情况下，通常可以忽略不计。

在分析反压时，有下面几种瓶颈可以检查：

• 系统资源
  首先，你应该检查执行机器的基本资源使用情况，如 CPU、网络或磁盘 I/O。如果某些资源被完全或大量使用，你可以采取以下措施之一：
  • 尝试优化代码。在这种情况下，代码剖析器很有帮助。
  • 基于上述特定的资源环境，调整Flink应用程序，以此来适配机器的资源使用情况。
  • 通过增加并行性和/或增加集群中的机器数量来扩大规模。
• GC垃圾回收
  通常，长时间的 GC 暂停会导致性能问题。您可以通过打印调试 GC 日志（通过 -XX:+PrintGCDetails）或使用一些内存/GC 剖析器来验证是否存在这种情况。由于处理 GC 问题在很大程度上取决于应用程序，且与 Flink 无关，在此不再赘述，有兴趣的可以参考Oracle 的《垃圾回收调整指南》或 Plumbr 的《Java 垃圾回收手册》。
• CPU/线程瓶颈
  有时，如果一个或几个线程造成了 CPU 瓶颈，从而导致整个机器的 CPU 使用率相对较低，CPU 瓶颈可能一眼看不出来。例如，在 48 核机器上，单个 CPU 瓶颈线程的 CPU 使用率仅为 2%。为此，可以考虑使用代码剖析器（Code Profilers），因为它们可以通过显示每个线程的 CPU 使用率来识别占用CPU过多的线程。
• 线程通信
  与上述 CPU/线程瓶颈问题类似，子任务也可能因共享资源上的线程竞争而出现瓶颈。在这方面，CPU Profilers 同样是你最好的助手！考虑在用户代码中查找同步开销/锁争用情况。还可以考虑调查共享系统资源。例如，JVM 的默认 SSL 实现可能会占用共享的 /dev/urandom 资源。
• 负载不均衡
  如果瓶颈是由数据偏斜造成的，可以尝试通过改变数据分区以打散热点key，或者预聚合来消除瓶颈或减轻其影响。

实际上远不止上述列出来的这些情况。一般来说，要减少背压，首先要分析瓶颈出现的位置，然后找出原因。开始推理 "原因 "的最佳途径是检查哪些资源得到了充分利用。

数据延迟监控

在本节中，我们将关注数据记录在 Flink应用程序流动时，其在 网络协议栈（包括系统的网络连接）内的等待时间。在吞吐量较低的情况下，这些延迟会直接受到缓冲区超时参数或间接受到应用程序代码处理延迟的影响。当处理记录的时间超过预期，或（多个）定时器同时启动并阻止接收器处理接收到的记录时，网络堆栈内记录的流动时间就会大大延长。因此，强烈建议在 Flink 作业中添加自己的指标，以便更好地跟踪作业组件中的数据延迟，并更广泛地了解延迟的原因。

目前，Flink 支持追踪记录流过应用系统（用户代码之外）的延迟。不过，默认情况下是禁用的（原因见下文）。该功能默认为禁用。要启用延迟追踪功能，必须在 Flink Configuration中将metrics.latency.interval 设置为正数（默认为0）或 ExecutionConfig#setLatencyTrackingInterval() 设置延迟跟踪间隔来启用。启用后，Flink 将根据通过 metrics.latency.granularity 定义的粒度来收集延迟直方图：

• single：每个操作符子任务对应一个直方图
• operator（default）：source任务和操作符子任务的每个组合都对应一个直方图
• subtask：source子任务和操作符子任务的每种组合都对应一个直方图。

这些指标是通过特殊的 “延迟标记” （latency markers）来收集的：每个source子任务都会周期性地发出一条特殊记录，其中包含其创建的时间戳。然后，“延迟标记”会与正常记录一起流动，但不会在线路上或缓冲队列内超越正常记录。同时，“延迟标记”并不会进入算子逻辑，并不包含记录在操作符中的处理时间。只有当操作符算子无法接收新记录，从而导致记录在缓存中排队时，使用Latency Markers测得的延迟才会反映出来。因此，延迟标记只能测量在操作符算子之间的等待时间，而不是完整的 "端到端 "延迟。不过，用户代码会间接影响这些等待时间！

由于 LatencyMarker 和普通记录一样位于网络缓冲区中，因此它们也会等待缓冲区满或因缓冲区超时而被刷新。当通道处于高负载时，网络缓冲数据不会增加延迟。但是，一旦一个通道处于低负载状态，记录和延迟标记就会出现最多为 buffer_timeout / 2 的预期平均延迟。 这种延迟会增加通向子任务的每个网络连接，在分析子任务的延迟指标时应将其考虑在内。

通过查看每个子任务暴露的延迟跟踪指标（latency tracking metrics）（例如第 95 百分位数），您应该能够确定哪些子任务大大增加了source-to-sink的总体延迟，并继续优化这些子任务。

总结

在本文中，我们讨论了如何监控 Flink 的网络协议栈，这主要涉及识别反压：反压发生的位置、来源以及可能的原因。这可以通过两种方式实现：

1. 使用反压监控器来观察简单场景以及调试；
2. 使用 Flink 的任务和网络堆栈指标进行持续监控、更深入的分析，并减少运行时开销。
   反压可能是由网络层本身造成的，但在大多数情况下，是由高负载下的某些子任务造成的。通过分析上述指标，可以将这两种情况区分开来。同时，Flink还提供了一些监控资源使用情况和跟踪网络延迟的指标。