Flink调优----资源配置调优与状态及Checkpoint调优

目录

第 1 章 资源配置调优

1.1 内存设置

1.1.1 TaskManager 内存模型

1、内存模型详解

2、案例分析

1.1.2 生产资源配置示例

1.2 合理利用 cpu 资源

1.2.1 使用 DefaultResourceCalculator 策略

1.2.2 使用 DominantResourceCalculator 策略

1.2.3 使用 DominantResourceCalculator 策略并指定容器 vcore 数

1.3 并行度设置

1.3.1 全局并行度计算

1.3.2 Source 端并行度的配置

1.3.3 Transform 端并行度的配置

1.3.4 Sink 端并行度的配置

第 2 章 状态及 Checkpoint 调优

2.1 RocksDB 大状态调优

2.1.1 开启 State 访问性能监控

2.1.2 开启增量检查点和本地恢复

1）开启增量检查点

2）开启本地恢复

3）设置多目录

2.1.3 调整预定义选项

2.1.4 增大 block 缓存

2.1.5 增大 write buffer 和 level 阈值大小

2.1.6 增大 write buffer 数量

2.1.7 增大后台线程数和 write buffer 合并数

1）增大线程数

2）增大 writebuffer 最小合并数

2.1.8 开启分区索引功能

2.1.9 参数设定案例

2.2 Checkpoint 设置

总结

---

       在大数据处理领域，Flink 作为一款强大的流处理框架，其性能优化对于高效数据处理至关重要。合理的资源配置是实现卓越性能的基石，它直接关系到 Flink 作业在处理大规模数据时的效率、稳定性以及资源利用率。而状态及 Checkpoint 调优则是确保数据处理准确性与可靠性的关键环节，能够有效应对系统故障与数据一致性挑战。通过深入探究资源配置调优以及状态和 Checkpoint 调优的策略与方法，可使 Flink 在复杂的数据处理场景中充分发挥其潜力，满足日益增长的实时数据处理需求，为企业和组织提供更加精准、及时的数据洞察与决策支持。

第 1 章 资源配置调优

        Flink 性能调优的第一步，就是为任务分配合适的资源，在一定范围内，增加资源的分配与性能的提升是成正比的，实现了最优的资源配置后，在此基础上再考虑进行后面论述的性能调优策略。

        提交方式主要是 yarn-per-job，资源的分配在使用脚本提交 Flink 任务时进行指定。

标准的 Flink 任务提交脚本（Generic CLI 模式）

从 1.11 开始，增加了通用客户端模式，参数使用 -D <property=value> 指定

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \ 指定并行度
-Dyarn.application.queue=test \ 指定 yarn 队列
-Djobmanager.memory.process.size=1024mb \ 指定 JM 的总进程大小
-Dtaskmanager.memory.process.size=1024mb \ 指定每个 TM 的总进程大小
-Dtaskmanager.numberOfTaskSlots=2 \ 指定每个 TM 的 slot 数
-c com.atguigu.flink.tuning.UvDemo \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar

参数列表：
//nightlies.apache.org/flink/flink-docs-release-1.13/docs/deployment/config/

1.1 内存设置

1.1.1 TaskManager 内存模型

1、内存模型详解

• JVM 特定内存：JVM本身使用的内存，包含JVM的metaspace和over-head

1）JVM metaspace：JVM元空间

taskmanager.memory.jvm-metaspace.size，默认256mb

2）JVM over-head执行开销：JVM执行时自身所需要的内容，包括线程堆栈、IO、编译缓存等所使用的内存。

taskmanager.memory.jvm-overhead.fraction，默认0.1

taskmanager.memory.jvm-overhead.min，默认192mb

taskmanager.memory.jvm-overhead.max，默认1gb

总进程内存*fraction，如果小于配置的min（或大于配置的max）大小，则使用min/max大小

• 框架内存：Flink框架，即TaskManager本身所占用的内存，不计入Slot的资源中。

堆内：taskmanager.memory.framework.heap.size，默认128MB

堆外：taskmanager.memory.framework.off-heap.size，默认128MB

• Task内存：Task执行用户代码时所使用的内存

堆内：taskmanager.memory.task.heap.size，默认none，由Flink内存扣除掉其他部分的内存得到。

堆外：taskmanager.memory.task.off-heap.size，默认0，表示不使用堆外内存

• 网络内存：网络数据交换所使用的堆外内存大小，如网络数据交换缓冲区

堆外：taskmanager.memory.network.fraction，默认0.1

  taskmanager.memory.network.min，默认64mb

  taskmanager.memory.network.max，默认1gb

Flink内存*fraction，如果小于配置的min（或大于配置的max）大小，则使用min/max大小

• 托管内存：用于RocksDB State Backend 的本地内存和批的排序、哈希表、缓存中间结果。

堆外：taskmanager.memory.managed.fraction，默认0.4

  taskmanager.memory.managed.size，默认none

如果size没指定，则等于Flink内存*fraction

2、案例分析

        基于 Yarn 模式，一般参数指定的是总进程内存，taskmanager.memory.process.size，比如指定为 4G，每一块内存得到大小如下：

（1）计算Flink内存

        JVM元空间256m

        JVM执行开销： 4g*0.1=409.6m，在[192m,1g]之间，最终结果409.6m

        Flink内存=4g-256m-409.6m=3430.4m

（2）网络内存=3430.4m*0.1=343.04m，在[64m,1g]之间，最终结果343.04m

（3）托管内存=3430.4m*0.4=1372.16m

（4）框架内存，堆内和堆外都是128m

（5）Task堆内内存=3430.4m-128m-128m-343.04m-1372.16m=1459.2m

所以进程内存给多大，每一部分内存需不需要调整，可以看内存的使用率来调整。

1.1.2 生产资源配置示例

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \ 指定并行度
-Dyarn.application.queue=test \ 指定 yarn 队列
-Djobmanager.memory.process.size=2048mb \ JM2~4G 足够
-Dtaskmanager.memory.process.size=4096mb \ 单个 TM2~8G 足够
-Dtaskmanager.numberOfTaskSlots=2 \ 与容器核数 1core：1slot 或 2core：1slot
-c com.atguigu.flink.tuning.UvDemo \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar

        Flink 是实时流处理，关键在于资源情况能不能抗住高峰时期每秒的数据量，通常用 QPS/TPS 来描述数据情况。

1.2 合理利用 cpu 资源

        Yarn 的容量调度器默认情况下是使用 “DefaultResourceCalculator” 分配策略，只根据内存调度资源，所以在 Yarn 的资源管理页面上看到每个容器的 vcore 个数还是 1。

        可以修改策略为 DominantResourceCalculator，该资源计算器在计算资源的时候会综合考虑 cpu 和内存的情况。在 capacity-scheduler.xml 中修改属性:

<property><name>yarn.scheduler.capacity.resource-calculator</name><!-- <value>org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator</value> --><value>org.apache.hadoop.yarn.util.resource.DominantResourceCalculator</value>
</property>

1.2.1 使用 DefaultResourceCalculator 策略

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \
-Drest.flamegraph.enabled=true \
-Dyarn.application.queue=test \
-Djobmanager.memory.process.size=1024mb \
-Dtaskmanager.memory.process.size=4096mb \
-Dtaskmanager.numberOfTaskSlots=2 \
-c com.atguigu.flink.tuning.UvDemo \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar

可以看到一个容器只有一个 vcore：

1.2.2 使用 DominantResourceCalculator 策略

修改后 yarn 配置后，分发配置并重启 yarn，再次提交 flink 作业：

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \
-Drest.flamegraph.enabled=true \
-Dyarn.application.queue=test \
-Djobmanager.memory.process.size=1024mb \
-Dtaskmanager.memory.process.size=4096mb \
-Dtaskmanager.numberOfTaskSlots=2 \
-c com.atguigu.flink.tuning.UvDemo \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar

看到容器的 vcore 数变了:

JobManager 1 个，占用 1 个容器，vcore = 1
TaskManager 3 个，占用 3 个容器，每个容器 vcore = 2，总 vcore = 2*3 = 6，因为默认单个容器的 vcore 数 = 单 TM 的 slot 数

1.2.3 使用 DominantResourceCalculator 策略并指定容器 vcore 数

指定 yarn 容器的 vcore 数，提交：

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \
-Drest.flamegraph.enabled=true \
-Dyarn.application.queue=test \
-DYarn.containers.vcores=3 \
-Djobmanager.memory.process.size=1024mb \
-Dtaskmanager.memory.process.size=4096mb \
-Dtaskmanager.numberOfTaskSlots=2 \
-c com.atguigu.flink.tuning.UvDemo \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar

 看到容器的 vcore 数变了:

JobManager 1 个，占用 1 个容器，vcore = 1
TaskManager 3 个，占用 3 个容器，每个容器 vcore = 3，总 vcore = 3*3 = 9

1.3 并行度设置

1.3.1 全局并行度计算

        开发完成后，先进行压测。任务并行度给 10 以下，测试单个并行度的处理上限。然后总 QPS / 单并行度的处理能力 = 并行度

        开发完 Flink 作业，压测的方式很简单，先在 kafka 中积压数据，之后开启 Flink 任务，出现反压，就是处理瓶颈。相当于水库先积水，一下子泄洪。

        不能只从 QPS 去得出并行度，因为有些字段少、逻辑简单的任务，单并行度一秒处理几万条数据。而有些数据字段多，处理逻辑复杂，单并行度一秒只能处理 1000 条数据。

        最好根据高峰期的 QPS 压测，并行度 * 1.2 倍，富余一些资源。

1.3.2 Source 端并行度的配置

        数据源端是 Kafka，Source 的并行度设置为 Kafka 对应 Topic 的分区数。

        如果已经等于 Kafka 的分区数，消费速度仍跟不上数据生产速度，考虑下 Kafka 要扩大分区，同时调大并行度等于分区数。

        Flink 的一个并行度可以处理一至多个分区的数据，如果并行度多于 Kafka 的分区数，那么就会造成有的并行度空闲，浪费资源。

1.3.3 Transform 端并行度的配置

• Keyby 之前的算子

        一般不会做太重的操作，都是比如 map、filter、flatmap 等处理较快的算子，并行度可以和 source 保持一致。

• Keyby 之后的算子

        如果并发较大，建议设置并行度为 2 的整数次幂，例如：128、256、512；
        小并发任务的并行度不一定需要设置成 2 的整数次幂；
        大并发任务如果没有 KeyBy，并行度也无需设置为 2 的整数次幂；

1.3.4 Sink 端并行度的配置

        Sink 端是数据流向下游的地方，可以根据 Sink 端的数据量及下游的服务抗压能力进行评估。如果 Sink 端是 Kafka，可以设为 Kafka 对应 Topic 的分区数。

        Sink 端的数据量小，比较常见的就是监控告警的场景，并行度可以设置的小一些。

        Source 端的数据量是最小的，拿到 Source 端流过来的数据后做了细粒度的拆分，数据量不断的增加，到 Sink 端的数据量就非常大。那么在 Sink 到下游的存储中间件的时候就需要提高并行度。

        另外 Sink 端要与下游的服务进行交互，并行度还得根据下游的服务抗压能力来设置，如果在 Flink Sink 这端的数据量过大的话，且 Sink 处并行度也设置的很大，但下游的服务完全撑不住这么大的并发写入，可能会造成下游服务直接被写挂，所以最终还是要在 Sink 处的并行度做一定的权衡。

第 2 章 状态及 Checkpoint 调优

2.1 RocksDB 大状态调优

        RocksDB 是基于 LSM Tree 实现的（类似 HBase），写数据都是先缓存到内存中，所以 RocksDB 的写请求效率比较高。RocksDB 使用内存结合磁盘的方式来存储数据，每次获取数据时，先从内存中 blockcache 中查找，如果内存中没有再去磁盘中查询。使用 RocksDB 时，状态大小仅受可用磁盘空间量的限制，性能瓶颈主要在于 RocksDB 对磁盘的读请求，每次读写操作都必须对数据进行反序列化或者序列化。当处理性能不够时，仅需要横向扩展并行度即可提高整个 Job 的吞吐量。

        从 Flink1.10 开始，Flink 默认将 RocksDB 的内存大小配置为每个 task slot 的托管内存。调试内存性能的问题主要是通过调整配置项 taskmanager.memory.managed.size 或者 taskmanager.memory.managed.fraction 以增加 Flink 的托管内存 (即堆外的托管内存)。进一步可以调整一些参数进行高级性能调优，这些参数也可以在应用程序中通过 RocksDBStateBackend.setRocksDBOptions (RocksDBOptionsFactory) 指定。下面介绍提高资源利用率的几个重要配置：

2.1.1 开启 State 访问性能监控

        Flink 1.13 中引入了 State 访问的性能监控，即 latency trackig state。此功能不局限于 State Backend 的类型，自定义实现的 State Backend 也可以复用此功能。

        State 访问的性能监控会产生一定的性能影响，所以，默认每 100 次做一次取样 (sample)，对不同的 State Backend 性能损失影响不同：
        对于 RocksDB State Backend，性能损失大概在 1% 左右
        对于 Heap State Backend，性能损失最多可达 10%

state.backend.latency-track.keyed-state-enabled：true #启用访问状态的性能监控
state.backend.latency-track.sample-interval: 100   #采样间隔
state.backend.latency-track.history-size: 128    #保留的采样数据个数，越大越精确
state.backend.latency-track.state-name-as-variable: true #将状态名作为变量

正常开启第一个参数即可。

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \
-Drest.flamegraph.enabled=true \
-Dyarn.application.queue=test \
-Djobmanager.memory.process.size=1024mb \
-Dtaskmanager.memory.process.size=4096mb \
-Dtaskmanager.numberOfTaskSlots=2 \
-Dstate.backend.latency-track.keyed-state-enabled=true \
-c com.atguigu.flink.tuning.RocksdbTuning \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar

2.1.2 开启增量检查点和本地恢复

1）开启增量检查点

        RocksDB 是目前唯一可用于支持有状态流处理应用程序增量检查点的状态后端，可以修改参数开启增量检查点：

state.backend.incremental: true   #默认 false，改为 true。

或代码中指定

new EmbeddedRocksDBStateBackend(true)

2）开启本地恢复

        当 Flink 任务失败时，可以基于本地的状态信息进行恢复任务，可能不需要从 hdfs 拉取数据。本地恢复目前仅涵盖键控类型的状态后端（RocksDB），MemoryStateBackend 不支持本地恢复并忽略此选项。

state.backend.local-recovery: true

3）设置多目录

如果有多块磁盘，也可以考虑指定本地多目录

state.backend.rocksdb.localdir:
/data1/flink/rocksdb,/data2/flink/rocksdb,/data3/flink/rocksdb

注意：不要配置单块磁盘的多个目录，务必将目录配置到多块不同的磁盘上，让多块磁盘来分担压力。

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \
-Drest.flamegraph.enabled=true \
-Dyarn.application.queue=test \
-Djobmanager.memory.process.size=1024mb \
-Dtaskmanager.memory.process.size=4096mb \
-Dtaskmanager.numberOfTaskSlots=2 \
-Dstate.backend.incremental=true \
-Dstate.backend.local-recovery=true \
-Dstate.backend.latency-track.keyed-state-enabled=true \
-c com.atguigu.flink.tuning.RocksdbTuning \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar

2.1.3 调整预定义选项

        Flink 针对不同的设置为 RocksDB 提供了一些预定义的选项集合，其中包含了后续提到的一些参数，如果调整预定义选项后还达不到预期，再去调整后面的 block、writebuffer 等参数。

        当前支持的预定义选项有 DEFAULT、SPINNING_DISK_OPTIMIZED、SPINNING_DISK_OPTIMIZED_HIGH_MEM 或 FLASH_SSD_OPTIMIZED。有条件上 SSD 的，可以指定为 FLASH_SSD_OPTIMIZED

state.backend.rocksdb.predefined-options： SPINNING_DISK_OPTIMIZED_HIGH_MEM 
#设置为机械硬盘+内存模式

2.1.4 增大 block 缓存

        整个 RocksDB 共享一个 block cache，读数据时内存的 cache 大小，该参数越大读数据时缓存命中率越高，默认大小为 8 MB，建议设置到 64 ~ 256 MB。

state.backend.rocksdb.block.cache-size: 64m   #默认 8m

2.1.5 增大 write buffer 和 level 阈值大小

        RocksDB 中，每个 State 使用一个 Column Family，每个 Column Family 使用独占的 write buffer，默认 64MB，建议调大。
        调整这个参数通常要适当增加 L1 层的大小阈值 max-size-level-base，默认 256m。该值太小会造成能存放的 SST 文件过少，层级变多造成查找困难，太大会造成文件过多，合并困难。建议设为 target_file_size_base（默认 64MB）的倍数，且不能太小，例如 5~10 倍，即 320~640MB。

state.backend.rocksdb.writebuffer.size: 128m
state.backend.rocksdb.compaction.level.max-size-level-base: 320m

2.1.6 增大 write buffer 数量

        每个 Column Family 对应的 writebuffer 最大数量，这实际上是内存中 “只读内存表 “的最大数量，默认值是 2。对于机械磁盘来说，如果内存足够大，可以调大到 5 左右

state.backend.rocksdb.writebuffer.count: 5

2.1.7 增大后台线程数和 write buffer 合并数

1）增大线程数

        用于后台 flush 和合并 sst 文件的线程数，默认为 1，建议调大，机械硬盘用户可以改为 4 等更大的值

state.backend.rocksdb.thread.num: 4

2）增大 writebuffer 最小合并数

        将数据从 writebuffer 中 flush 到磁盘时，需要合并的 writebuffer 最小数量，默认值为 1，可以调成 3。

state.backend.rocksdb.writebuffer.number-to-merge: 3

2.1.8 开启分区索引功能

        Flink 1.13 中对 RocksDB 增加了分区索引功能，复用了 RocksDB 的 partitioned Index & filter 功能，简单来说就是对 RocksDB 的 partitioned Index 做了多级索引。也就是将内存中的最上层常驻，下层根据需要再 load 回来，这样就大大降低了数据 Swap 竞争。线上测试中，相对于内存比较小的场景中，性能提升 10 倍左右。如果在内存管控下 Rocksdb 性能不如预期的话，这也能成为一个性能优化点。

state.backend.rocksdb.memory.partitioned-index-filters:true  #默认 false

2.1.9 参数设定案例

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \
-Drest.flamegraph.enabled=true \
-Dyarn.application.queue=test \
-Djobmanager.memory.process.size=1024mb \
-Dtaskmanager.memory.process.size=4096mb \
-Dtaskmanager.numberOfTaskSlots=2 \
-Dstate.backend.incremental=true \
-Dstate.backend.local-recovery=true \
-Dstate.backend.rocksdb.predefined-options=SPINNING_DISK_OPTIMIZED_HIGH_MEM \
-Dstate.backend.rocksdb.block.cache-size=64m \
-Dstate.backend.rocksdb.writebuffer.size=128m \
-Dstate.backend.rocksdb.compaction.level.max-size-level-base=320m \
-Dstate.backend.rocksdb.writebuffer.count=5 \
-Dstate.backend.rocksdb.thread.num=4 \
-Dstate.backend.rocksdb.writebuffer.number-to-merge=3 \
-Dstate.backend.rocksdb.memory.partitioned-index-filters=true \
-Dstate.backend.latency-track.keyed-state-enabled=true \
-c com.atguigu.flink.tuning.RocksdbTuning \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar

2.2 Checkpoint 设置

        一般需求，我们的 Checkpoint 时间间隔可以设置为分钟级别（1 ~5 分钟）。对于状态很大的任务每次 Checkpoint 访问 HDFS 比较耗时，可以设置为 5~10 分钟一次 Checkpoint，并且调大两次 Checkpoint 之间的暂停间隔，例如设置两次 Checkpoint 之间至少暂停 4 或 8 分钟。同时，也需要考虑时效性的要求，需要在时效性和性能之间做一个平衡，如果时效性要求高，结合 end-to-end 时长，设置秒级或毫秒级。如果 Checkpoint 语义配置为 EXACTLY_ONCE，那么在 Checkpoint 过程中还会存在 barrier 对齐的过程，可以通过 Flink Web UI 的 Checkpoint 选项卡来查看 Checkpoint 过程中各阶段的耗时情况，从而确定到底是哪个阶段导致 Checkpoint 时间过长然后针对性的解决问题。

        RocksDB 相关参数在前面已说明，可以在 flink-conf.yaml 指定，也可以在 Job 的代码中调用 API 单独指定，这里不再列出。

// 使⽤ RocksDBStateBackend 做为状态后端，并开启增量 Checkpoint
RocksDBStateBackend rocksDBStateBackend = new RocksDBStateBackend("hdfs://hadoop1:8020/flink/checkpoints", true);
env.setStateBackend(rocksDBStateBackend);// 开启 Checkpoint，间隔为 3 分钟
env.enableCheckpointing(TimeUnit.MINUTES.toMillis(3));
// 配置 Checkpoint
CheckpointConfig checkpointConf = env.getCheckpointConfig();
checkpointConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 最小间隔 4 分钟
checkpointConf.setMinPauseBetweenCheckpoints(TimeUnit.MINUTES.toMillis(4));
// 超时时间 10 分钟
checkpointConf.setCheckpointTimeout(TimeUnit.MINUTES.toMillis(10));
// 保存 checkpoint
checkpointConf.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

总结

        本文全面且深入地阐述了 Flink 性能调优中的资源配置调优以及状态及 Checkpoint 调优两大部分内容。在资源配置调优方面，涵盖了内存设置，包括对 JVM 特定内存、框架内存、Task 内存、网络内存和托管内存的详细剖析，并给出基于 Yarn 模式的案例分析与生产资源配置示例；介绍了合理利用 CPU 资源的策略，如通过修改 Yarn 容量调度器的资源计算器策略来综合考虑 CPU 和内存；还阐述了并行度设置的方法，包括全局并行度计算以及 Source、Transform、Sink 端的并行度设置要点。在状态及 Checkpoint 调优部分，针对 RocksDB 大状态调优，提出了如开启 State 访问性能监控、增量检查点和本地恢复、调整预定义选项、增大 block 缓存、write buffer 及相关参数、开启分区索引功能等一系列措施，并给出参数设定案例；同时对 Checkpoint 设置进行了说明，包括时间间隔、暂停间隔的设置以及在不同时效性要求下的考量，还提及了如何通过 Flink Web UI 排查 Checkpoint 耗时问题。通过对这些调优策略的掌握与运用，能够显著提升 Flink 作业的性能、可靠性与资源利用效率，助力在大数据处理任务中取得更优的成果。