Flink 如何处理背压

文章目录

前言

一、什么是背压？

二、处理背压的步骤

1.模拟背压机制

2.为什么要关心背压问题？

总结

前言

初次接触Flink的同学会对背压有很多的疑问。本文就是我学习的一些心得和体会，以及借鉴一些文章的感想。

Flink 如何处理背压效应。

答案很简单：Flink 没有使用任何复杂的机制，因为它不需要。由于是纯数据流引擎，它可以优雅地响应背压。

一、什么是背压？

背压是指系统接收数据的速率高于它在临时负载高峰期间可以处理的速率的情况。

消息处理速度 < 消息的发送速度，这意味着Sink消耗数据的速度比Source生成速度慢。Sink正在向上游算子施加压力Source。消息拥堵，系统运行不畅。

在工作流中数据记录是从上游向下游流动的（例如：从 Source 到 Sink）。反压沿着相反的方向传播，沿着数据流向上游传播。

如果你看到 subtasks 的状态为 OK 表示没有反压。HIGH 表示这个 subtask 被反压。状态用如下定义：

OK: 0% <= 反压比例 <= 10%
LOW: 10% < 反压比例 <= 50%
HIGH: 50% < 反压比例 <= 100%

WebUI 集合了所有 subTasks 的反压和繁忙指标的最大值，并在 JobGraph 中将集合的值进行显示。除了显示原始的数值，tasks 也用颜色进行了标记，使检查更加容易。

闲置的 tasks 为蓝色，完全被反压的 tasks 为黑色，完全繁忙的 tasks 被标记为红色。中间的所有值都表示为这三种颜色之间的过渡色。

Task 性能指标 #

Task（SubTask）的每个并行实例都可以用三个一组的指标评价：

backPressureTimeMsPerSecond，subtask 被反压的时间
idleTimeMsPerSecond，subtask 等待某类处理的时间
busyTimeMsPerSecond，subtask 实际工作时间在任何时间点，这三个指标相加都约等于1000ms。

这些指标每两秒更新一次，上报的值表示 subtask 在最近两秒被反压（或闲或忙）的平均时长。当你的工作负荷是变化的时需要尤其引起注意。比如，一个以恒定50%负载工作的 subtask 和另一个每秒钟在满负载和闲置切换的 subtask 的busyTimeMsPerSecond值相同，都是500ms。

在内部，反压根据输出 buffers 的可用性来进行判断的。如果一个 task 没有可用的输出 buffers，那么这个 task 就被认定是在被反压。相反，如果有可用的输入，则可认定为闲置，

可以得出: 第三个算子的处理数据速度比第二个算子生成数据的速度, 明显的解决方法: 提高第三个算子的并发度, 问题又出现了: 并发度要上调到多少呢?

第一次上调, 从原来的10并发上调到 40

观察缓存池对比的情况:

并发是10的buffer情况: (背压的情况比较严重, 曲线持续性地达到峰值, 会导致资源占光)

并发是40的buffer情况:(有了比较大的改善, 但是还是存在背压的问题, 因为曲线有达到顶峰的时候)

并发是100的buffer情况: (背压的情况已经大大缓解)

许多日常情况都会导致背压。例如，垃圾收集停顿会导致传入数据堆积，或者数据源可能会出现发送数据速度的峰值（类似于kafka中的数据洪峰的概念）。背压如果处理不当，可能会导致资源耗尽，甚至在最坏的情况下会导致数据丢失（因此大数据处理框架一定要面临和处理此问题），对于许多流应用程序来说，数据丢失是不可接受的，这些应用程序需要对记录进行一次处理。

额外的数据需要缓冲在某处。缓冲也应该是持久的，因为在出现故障的情况下，需要重放这些数据以防止数据丢失。理想情况下，这些数据应该缓存在一个持久通道中（例如，如果源保证持久性，那么源本身就是这种通道的一个主要例子）。理想的反应是对从 sink 到 source 的整个管道进行“背压”，并对 source 进行节流，以将速度调整到管道最慢部分的速度，达到稳定状态：

二、处理背压的步骤

1.模拟背压机制

Flink 运行时的构建块是操作符和流。每个操作符都在使用中间流，通过对它们进行转换，产生新的流。描述网络机制的最佳类比是 Flink 使用具有有限容量的有效分布式阻塞队列。与 Java 连接线程的常规阻塞队列一样，一旦队列的缓冲区耗尽（有界容量），处理速度较慢的接收器就会降低发送器发送数据的速度。

2.为什么要关心背压问题？

背压是你的机器或Operator过载的指标。背压的积累直接影响系统的端到端延迟，因为记录在被处理之前在队列中等待的时间更长。其次，对齐检查点在背压下需要更长的时间，而未对齐的检查点会更大（您可以在文档中阅读有关对齐和未对齐检查点的更多信息。如果您在检查点屏障传播时间方面苦苦挣扎，那么处理背压很可能有助于解决问题问题。最后，您可能只想优化您的作业以降低运行作业的成本。

为了解决所有情况下的问题，需要意识到它，然后定位和分析它。

3.如何处理背压？

假设你确定了背压的来源，也就是瓶颈所在，下一步就是分析为什么会发生这种情况。下面我们按照从基本到复杂的顺序列出了导致背压的一些潜在成因。我们建议首先检查基本成因，然后再深入研究更复杂的成因，否则就可能得出一些错误的结论。
另外回想一下，背压可能是暂时的，可能是由于负载高峰、检查点或作业重启时数据 backlog 待处理导致的结果。如果背压是暂时的，那么忽略它就行了。此外还要记住，分析和解决问题的过程可能会受到瓶颈本身的影响。话虽如此，这里还是有几件事需要检查一下。

系统资源

首先，你应该检查受控机器的基本资源使用情况，如 CPU、网络或磁盘 I/O 等指标。如果某些资源在被全部或大量占用，你可以执行以下操作：

尝试优化你的代码。此时代码分析器是很有用的。
调整这项Flink task的资源。
通过增加并行度和 / 或增加群集中的计算机数量来扩展资源。

垃圾收集

一般来说，长时间的垃圾回收工作会引发性能问题。你可以打印 GC 调试日志（通过 -XX: +PrintGCDetails）或使用某些内存 /GC 分析器来验证你是否处于这种状况下。由于 GC 问题的处理与应用程序高度相关，并且独立于 Flink，因此我们不会在此详细介绍（可参考 Oracle 的垃圾收集调整指南，https://docs.oracle.com/javase/8/docs/technotes/guides/vm/gctuning/index.html 或 Plumbr 的 Java 垃圾回收手册，Splunk® Application Performance Monitoring | Splunk）。

CPU 线程瓶颈

如果 CPU 瓶颈来自于一个或几个线程，而整台机器的 CPU 使用率仍然相对较低，则 CPU 瓶颈可能就很难被发现了。例如，48 核计算机上的单个 CPU 线程瓶颈只会带来 2％的 CPU 使用率。可以考虑使用代码分析器，因为它们可以显示每个线程的 CPU 使用情况，这样就能识别出热线程。

线程争用

与上面的 CPU 线程瓶颈问题类似，共享资源上较高的线程争用率可能会导致子任务瓶颈。还是要请出 CPU 分析器，考虑查找用户代码中的同步开销 / 锁争用——虽然我们应该避免在用户代码中添加同步性，这可能很危险！还可以考虑调查共享系统资源。例如，默认 JVM 的 SSL 实现可以从共享的 /dev/urandom 资源周围获取数据。

加载不均衡

如果你的瓶颈是由数据偏差【数据倾斜】引起的，可以尝试将数据分区更改为几个独立的重键，或实现本地 / 预聚合来清除偏差或减轻其影响。

除此之外还有很多情况。一般来说，为了削弱瓶颈从而减少背压，首先要分析它发生的位置，然后找出原因。最好从检查哪些资源处于充分利用状态开始入手。

延迟追踪

追踪各个可能环节出现的延迟是一个独立的话题。在本文中，我们将重点关注 Flink 网络栈中的记录的等待时间——包括系统网络连接的情况。在吞吐量较低时，这些延迟会直接受输出刷新器的缓存超时参数的影响，或间接受任何应用程序代码延迟的影响。处理记录的时间比预期的要长或者（多个）计时器同时触发，并阻止接收器处理传入的记录时，网络栈内后续记录的等待时间会大大延长。强烈建议开发人员将自己的指标添加到 Flink 作业中，以便更好地跟踪作业组件中的延迟，并更全面地了解延迟产生的原因。

Flink 为追踪通过系统（用户代码之外）的记录延迟提供了一些支持。但默认情况下此功能被禁用（原因参见下文！），必须用 metrics.latency.interval 或 ExecutionConfig #setLatencyTrackingInterval() 在 Flink 的配置中设置延迟追踪间隔才能启用此功能。启用后，Flink 将根据 metrics.latency.granularity 定义的粒度生成延迟直方图：

single：每个操作符子任务有一个直方图
operator（默认值）：源任务和操作符子任务的每个组合有一个直方图
subtask：源子任务和操作符子任务的每个组合有一个直方图（并行度翻了两番！）

这些指标通过特殊的“延迟标记”收集：每个源子任务将定期发出包含其创建时间戳的特殊记录。然后，延迟标记与正常记录一起流动，不会在线路上或缓存队列中超过正常记录。但是，延迟标记不会进入应用程序逻辑，并会在那里超过正常记录。因此，延迟标记仅测量用户代码之间的等待时间，而不是完整的“端到端”延迟。但用户代码会间接影响这些等待时间！

由于 LatencyMarker 就像普通记录一样位于网络缓冲区中，它们也会因缓存已满而等待，或因缓存超时而刷新。当信道处于高负载时，网络缓冲区数据不会增加延迟。但是只要一个信道处于低负载状态，记录和延迟标记就会承受最多 buffer_timeout/2 的平均延迟。这个延迟会加到每个连接子任务的网络连接上，在分析子任务的延迟指标时应该考虑这一点。

只要查看每个子任务暴露的延迟追踪指标，例如在第 95 百分位，你就应该能识别出是哪些子任务在显著影响源到汇延迟，然后对其做针对性优化。