摘要：本文整理自俞航翔、陈婧敏、黄鹏程老师所撰写的大状态作业调优实践指南。由于内容丰富，本文分享上篇内容，主要分为以下四个部分：
- Flink 状态（State）简介
- 大状态作业导致的问题
- 大状态作业诊断调优整体思路
- Flink Datastream 作业大状态导致反压的调优原理与方法
一、Flink 状态（State）简介
Apache Flink 是一个开源的流处理框架，用于处理和分析实时数据流。在 Flink 中，状态管理是流处理应用的核心概念之一，它允许算子（operators）在处理事件时保持和操作状态信息。 在 Flink 中，状态可以被视为算子的“记忆”，它使得算子能够在处理无界流数据时保持对历史数据的跟踪。状态可以是简单的键值对，也可以是更复杂的数据结构，如列表、集合或自定义对象。状态的更新和查询对于实现复杂的流处理逻辑至关重要。
具体说来，首先，Checkpoint 是 Flink 的一种容错机制。它通过周期性地自动保存作业状态到一个持久化存储系统中，来确保在发生故障时可以从最后一个成功的 Checkpoint 恢复作业状态，从而保证数据处理的一致性和准确性。Checkpoint 的触发是自动的，并且对用户是透明的，不需要用户进行额外的编程工作。 当 Flink 作业正在运行时，Checkpoint 机制会定期激活，触发作业的状态（如中间结果、配置信息等）被保存到预先定义的存储位置。这些状态信息被保存后，如果发生故障，Flink 可以使用这些保存的状态信息来重建作业的状态，从而恢复作业的正常运行。
而Savepoint 与 Checkpoint 类似，也是 Flink 用于保存作业状态的机制。不同之处在于，Savepoint 是显式的，需要用户手动触发。用户可以在任何时刻创建一个 Savepoint，以便在需要时使用这个状态来重启作业。Savepoint 为用户提供了更多的灵活性和控制权，因为它允许用户在特定的时刻创建作业的快照。 Savepoint 可以用于作业升级、回滚或者恢复到一个已知的稳定状态，这对于作业的维护和更新非常有用。与 Checkpoint 一样，Savepoint 也是 Flink 容错机制的一部分，它们共同确保了 Flink 作业的高可用性和稳定性。
状态管理与使用是阿里云实时计算Flink版中的重要功能，通过产品的控制台可以完成检查点生命周期的自动管理，并在保证不影响作业可用性的前提下最小化存储空间，同时透出了快照的管理和共享机制，提供了用户不同场景下的快照选择，而作业间的快照共享功能对大状态作业的 A/B Test 和主备链路切换具备极为实用的价值。
二、大状态作业导致的问题
在处理大规模状态作业的过程中，系统面临着调优的严峻挑战。随着作业状态的持续膨胀，多个问题逐步显现，对作业的整体性能产生不利影响：
- 性能下降与作业反压：随着有状态算子状态的累积，I/O资源的瓶颈问题日益凸显，引发作业反压。这不仅增加了处理延迟，还导致吞吐量（TPS）降低。
- 资源利用效率低下：有状态算子的CPU资源常出现大量闲置，且随着状态规模的增长，资源浪费问题更加严重。
- 检查点与快照机制的时效性问题：状态规模的扩大使得检查点和快照过程更易超时，这不仅增加了作业重启后追赶数据的时间成本，也对端到端的 Exactly-once 语义的实现带来了额外延迟。
- 启动与扩缩容过程缓慢：在作业启动和扩缩容过程中，每个算子节点需从全量数据中恢复并重建本地数据库，这一过程的时间消耗与状态规模成正比。拥有大状态作业的状态加载往往成为启动和扩缩容执行速度的瓶颈，进而延长业务中断时间。
三、大状态作业诊断调优整体思路
在处理 Flink 作业时，如下这三类问题通常由大规模状态的管理和维护所引起：运行时性能减缓、检查点或快照超时问题以及作业启动和扩缩容过程缓慢。为了优化这些大状态作业，建议遵循以下步骤：
识别作业瓶颈：通过诊断工具结合具体业务产出情况，对作业目前的运行情况进行更为深入的了解，进而确定作业的性能瓶颈是否与状态管理有关。
采用更新的引擎版本：Flink在状态模块持续优化，最新版本的引擎通常具有更高的性能。阿里云实时计算的Flink企业版——Ververica Runtime (VVR)，与 Apache Flink 完全兼容，并内置了专为流计算优化的状态存储 Gemini。Gemini 针对状态访问进行了设计，有效提升了性能、检查点和作业恢复能力，且参数自适应，无需手动配置。结合实时计算产品，VVR 为用户提供了企业级的优化体验，确保性能达到最佳。在进行性能调优前，请确保已采用最新版引擎和相关配置。
-
针对不同问题采取特定调优策略：
（1）运行时性能下降（作业反压）：在这种情况下，调优应遵循以下顺序：首先优化SQL层，其次基于TTL（生存时间）减少状态数据，然后调整内存和并发资源以降低磁盘读取频率。
（2）检查点或快照超时：在处理此类问题时，应先优化作业的运行时性能以减轻反压，接着优化同步阶段的性能，然后调整并发资源以降低单个并发任务的状态量，最后考虑使用原生快照功能来提高效率。
（3）作业启动和扩缩容缓慢：如果本地磁盘资源充足，可以优先考虑启用状态本地恢复（Local Recovery）功能。同时，利用 Gemini 的懒加载特性和延迟剪裁技术，可以有效提升作业的启动和扩缩容速度。
四、Flink Datastream 作业大状态导致反压的调优原理与方法
如“大状态作业导致的问题”一节中所述，状态管理不仅影响应用的性能，还关系到系统的稳定性和资源的有效利用。如果状态管理不当，可能会导致性能下降、资源耗尽，甚至系统崩溃。而 Flink Datastream API 为用户操作状态提供了非常灵活的接口，因此开发者需要采取一系列措施来确保状态大小可控，避免状态的无限制增长。
1. 基本原理
Flink 支持 Operator State 和 Keyed State 两种状态，其中大状态问题通常由 Keyed State 引起。Flink Datastream API 支持通过显式的ValueState、ListState、MapState等状态接口来维护 Keyed State，以及为其设置过期时间，详情可参考相关介绍[11]【状态相关介绍】
2. 问题诊断方法
在Flink作业遭遇性能瓶颈时，系统往往表现出明显的反压现象。这种反压可能由多种因素引起，但主要的原因之一是作业状态规模的持续膨胀，直至超出内存限制。此时，状态存储引擎不得不将部分不频繁使用的状态数据移至磁盘，而磁盘与内存在数据存取速度上的巨大差异，使得磁盘 IO 操作成为数据处理效率的瓶颈。尤其在 Flink 的计算过程中，如果算子频繁地从磁盘读取状态数据，将显著增加作业的延迟，降低整体处理速度，成为性能问题的根源。
为了准确识别是否由状态访问引发反压，我们需要对作业的运行状态和算子行为进行深入分析。利用监控工具追踪和诊断性能瓶颈，可以有效地发现并解决由状态访问引起的性能问题，从而提升 Flink 作业的性能。
阿里云 Flink 平台集成了多种分析工具，如 Thread Dump、线程动态分析、火焰图以及 State & Checkpoint 相关指标等，结合智能诊断与自动调优功能，形成了初步的大状态作业问题诊断与自动调优能力。这些工具在定位和解决大规模状态作业的性能瓶颈方面发挥着重要作用。
| 工具 | 用途 | 使用方式 |
|---|---|---|
| Thread Dump | 查看当前时刻的 Operator 线程是否主要在访问 State | 在 Thread Dump 页面，按 Operator 名字搜索，观察线程栈是否在访问 State （线程栈持续在 Gemini 或 RocksDB 的访问链路上），操作示例见 “Thread Dump 使用方式” |
| 线程动态 | 抽样查看一段时间内的线程动态是否主要在访问 State | 在线程动态页面，按 Operator 名字搜索，采样一段时间，并观察线程栈是否在访问 State （线程栈持续在 Gemini 或 RocksDB 的访问链路上），操作示例见 “线程动态使用方式” |
| CPU 火焰图 | 查看一段时间内的 CPU 时间占比大头是否在访问 State | 在 CPU 火焰图页面，采样一段时间，并观察最长几条 CPU 链路上，方法是否在访问 State （方法在 Gemini 或 RocksDB 的访问链路上），操作示例见 “火焰图使用方式” |
| 查看运行监控指标 | 通过查看 State Size 相关指标判断 状态大小和 IO 情况 | 在指标页面，重点观察以下指标：State Size (Gemini Only): 运行时单并发的状态大小；lastCheckpointFullSize: 最近一个 Checkpoint 的全量大小，可以用来大致估算整体作业的状态大小；State Access Latency （需要额外开启）：当单个 State 访问达到毫秒级，需要重点关注下状态访问的性能操作示例见 “指标使用方式”。 |
2.1 Thread Dump 使用方式
（1）点击瓶颈算子进入TaskManager 性能查看页面，同时记录在 Detail 页面中的算子名
（2）进入 Thread Dump 页面 并按 1 中的瓶颈算子名搜索其线程栈，如下即是 Gemini State 访问的线程栈
2.2 线程动态使用方式
（1）类似上一节，点击瓶颈算子进入TaskManager 性能查看页面，同时记录在 Detail 页面中的算子名
（2）进入线程动态页面并按 1 中的瓶颈算子名搜索其线程栈并采样一段时间，观察其线程栈，如下即可观察到 Gemini State 访问的线程栈
2.3 火焰图使用方式
（1）类似上一节，点击瓶颈算子进入TaskManager 性能查看页面，同时记录在 Detail 页面中的算子名
（2）进入火焰图页面并观察占据 CPU 时间最长的方法，如下可以观察到有较多 Gemini State 访问的方法
2.4 指标使用方式
（1）进入指标页面
（2）可以在 State 和 Checkpoint 两部分观察上述的一些指标
3. 调优方法
3.1 反复确认业务逻辑，合理设计状态
在使用Flink进行状态管理时，首先需要审视业务逻辑，确保只存储必要的数据，避免产生不必要的状态信息。合理设计状态结构和存储内容是控制状态增长的关键所在。仅存储业务所需的最小化状态信息，有利于避免状态的无限增长。
设置合理状态生命周期减小状态大小
Flink 提供了丰富的状态时间特性，如 ValueStateDescriptor 的 setTTL 方法，可以设置状态的生命周期，确保状态在一定时间后自动过期并被清除。同时，开发者也可以直接调用 clear() 或 remove() 方法，显式删除不再需要的状态条目。合理利用这些特性，可以有效控制状态规模。
3.2 使用定时器进行状态清理
除了依赖状态的时间特性，开发者还可以利用 Flink 的定时器机制，定期触发状态的清理操作。通过设置合理的定时器触发时间，可以确保过期状态及时被清理，避免状态无限增长。这种主动清理状态的方法，可以更精细地控制状态的生命周期。
3.3 进行必要的监控与日志输出，同时定期分析状态文件
在状态管理过程中，需要持续监控状态大小和状态后端的性能指标，及时发现异常情况。同时，记录详细的日志信息，有助于在出现问题时快速定位和解决。除此之外，定期分析状态文件，也能够提供系统运行的历史数据，有助于识别作业模式和预测可能的风险点，为进一步优化状态管理提供依据。
3.4 尽可能减少读盘
为了提升系统性能，我们可以通过减少磁盘读取次数并优化内存使用来实现。以下是针对不同情况的具体策略：
（1）优化内存分配：在保证系统总资源不受影响的前提下，我们可以重新分配内存资源，将更多的内存分配给托管内存（Managed Memory）。这样做可以有效提升内存的命中率，从而减少对磁盘的依赖。具体操作时，应确保其他内存资源充足，以免影响系统的其他部分。
（2）细粒度资源配置：在进行资源配置时，应优先考虑增加内存资源。通过为存储引擎分配更多的托管内存，我们可以进一步提高内存命中率，减少对磁盘的读取需求。这种方法在细粒度的资源管理中尤为重要，因为它允许我们更精确地控制资源分配，以达到最佳的性能表现。
（3）提高并发处理能力：通过增加并发处理的数量，我们可以降低单个并发任务的状态量，从而减少需要写入磁盘的数据量。这种方法可以有效地减少磁盘 I/O 操作，提高整体的数据处理效率。
| 使用场景 | 方案 | 实践方式 | 注意事项 |
|---|---|---|---|
| 当 Heap 等其他内存资源余量较多时 | 调整内存资源比例，提供更多内存资源给 Managed Memory | 运行参数中配置参数：taskmanager.memory.managed.fraction[6]该参数默认值是 0.4，适当增加该参数可以将更多内存资源用于实际状态数据存储 | 需要确保其他内存资源够用，否则会导致 Full GC 频繁从而性能下降 |
| 所有场景 | 增加内存 | 在资源配置或细粒度资源配置中增加内存[7] | |
| 增加并发 | 在资源配置或细粒度资源配置中增加并发[7] |
参考文献
[6] taskmanager.memory.managed.fraction 配置说明
[7] 作业资源配置方式
[11] 状态相关介绍
