Flink调优总结

public class DataGenTest {     public static void main(String[] args) throws Exception{         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();         env.setParallelism(1);         SingleOutputStreamOperator<OrderInfo> orderInfoDS = env.addSource(new DataGeneratorSource<>(new RandomGenerator<OrderInfo>() {             @Override             public OrderInfo next() {                 return OrderInfo.builder()                         .id( random.nextInt(1, 100000))                         .userId( random.nextInt(1, 100000))                         .totalAmount(random.nextUniform(1, 1000))                         .createTime(System.currentTimeMillis())                         .build();             }         })).returns(Types.POJO(OrderInfo.class));         orderInfoDS.print("orderInfoDS");         env.execute("");     } }

public class DataGenSQLTest {     public static void main(String[] args) {         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();         env.setParallelism(1);         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);         String orderSql="CREATE TABLE order_info (\n" +                 " id INT,\n" +                 " userId BIGINT,\n" +                 " totalAmount DOUBLE,\n" +                 " createTime AS localtimestamp,\n" +                 " WATERMARK FOR createTime AS createTime\n" +                 ") WITH (\n" +                 " 'connector' = 'datagen',\n" +                 " 'rows-per-second'='20000',\n" +                 " 'fields.id.kind'='sequence',\n" +                 " 'fields.id.start'='1',\n" +                 " 'fields.id.end'='100000000',\n" +                 " 'fields.user_id.kind'='random',\n" +                 " 'fields.user_id.min'='1',\n" +                 " 'fields.user_id.max'='1000000',\n" +                 " 'fields.total_amount.kind'='random',\n" +                 " 'fields.total_amount.min'='1',\n" +                 " 'fields.total_amount.max'='1000'\n" +                 ")";         tableEnv.executeSql(orderSql);         tableEnv.executeSql("select * from order_info").print();     } }

metrics.latency.granularity: operator #默认 operator

single：每个算子单独统计延迟

operator（默认值）：每个下游算子都统计自己与 Source 算子之间的延迟

subtask：每个下游算子的 sub-task 都统计自己与 Source 算子的 sub-task 之间 的延迟

微批处理通过增加延迟换取高吞吐，如果有超低延迟的要求，不建议开启微批处理。通 常对于聚合的场景，微批处理可以显著的提升系统性能，建议开启

LocalGlobal 优 化 将 原 先 的 Aggregate 分 成 Local+Global 两 阶 段 聚 合 ， 即 MapReduce 模型中的 Combine+Reduce 处理模式。第一阶段在上游节点本地攒一批数据 进行聚合（localAgg），并输出这次微批的增量值（Accumulator）。第二阶段再将收到 的 Accumulator 合并（Merge），得到最终的结果（GlobalAgg）

1）LocalGlobal 优化需要先开启 MiniBatch，依赖于 MiniBatch 的参数

2）table.optimizer.agg-phase-strategy: 聚合策略。默认 AUTO，支持参数 AUTO、 TWO_PHASE(使用 LocalGlobal 两阶段聚合)、ONE_PHASE(仅使用 Global 一阶段聚合)。

从 Flink1.9.0 版 本 开 始 ， 提 供 了 COUNT DISTINCT 自 动 打 散 功 能 ， 通 过
HASH_CODE(distinct_key) % BUCKET_NUM 打散

table.optimizer.distinct-agg.split.enabled: true，默认 false。 table.optimizer.distinct-agg.split.bucket-num: Split Distinct 优化在第一层聚合中，被打散的 bucket 数目。默认 1024。

（1）目前不能在包含 UDAF 的 Flink SQL 中使用 Split Distinct 优化方法。 （2）拆分出来的两个 GROUP 聚合还可参与 LocalGlobal 优化。 （3）该功能在 Flink1.9.0 版本及以上版本才支持。

SELECT a, COUNT(DISTINCT b) AS total_b, COUNT(DISTINCT CASE WHEN c IN ('A', 'B') THEN b ELSE NULL END) AS AB_b, COUNT(DISTINCT CASE WHEN c IN ('C', 'D') THEN b ELSE NULL END) AS CD_b FROM T GROUP BY a

SELECT a, COUNT(DISTINCT b) AS total_b, COUNT(DISTINCT b) FILTER (WHERE c IN ('A', 'B')) AS AB_b, COUNT(DISTINCT b) FILTER (WHERE c IN ('C', 'D')) AS CD_b FROM T GROUP BY a

Flink SQL 优化器可以识别同一唯一键上的不同 FILTER 参数。如，在上面的示例中，三个 COUNT DISTINCT 都作在 b 列上。
此时，经过优化器识别后，Flink 可以只使用一个共享状态实例，而不是三个状态实例，可减少状态的大小和对状态的访问

如果 Checkpoint 做的非常慢，超过了 timeout 还没有完成，则整个 Checkpoint  也会失败

内存组成架构

案例分析

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.flink.tuning.UvDemo \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar

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

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

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.flink.tuning.UvDemo \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar

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

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.flink.tuning.UvDemo \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar

JobManager1 个，占用 1 个容器，vcore=1 TaskManager3 个，占用 3 个容器，每个容器 vcore =3，总 vcore=3*3=9

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

不能只从 QPS 去得出并行度，因为有些字段少、逻辑简单的任务，单并行度一秒处理
几万条数据。而有些数据字段多，处理逻辑复杂，单并行度一秒只能处理 1000 条数据。
最好根据高峰期的 QPS 压测，并行度*1.2 倍，富余一些资源

数据源端是 Kafka，Source 的并行度设置为 Kafka 对应 Topic 的分区数。
如果已经等于 Kafka 的分区数，消费速度仍跟不上数据生产速度，考虑下 Kafka 要扩
大分区，同时调大并行度等于分区数。

Keyby 之前的算子

Keyby 之后的算子

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

另外 Sink 端要与下游的服务进行交互，并行度还得根据下游的服务抗压能力来设置

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 #将状态名作为变量

开启增量检查点

开启本地恢复

设置多目录

当 前 支 持 的 预 定 义 选 项 有 DEFAULT 、 SPINNING_DISK_OPTIMIZED 、 SPINNING_DISK_OPTIMIZED_HIGH_MEM 或 FLASH_SSD_OPTIMIZED

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

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

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

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

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

state.backend.rocksdb.writebuffer.count: 5

增大线程数

增大 writebuffer 最小合并数

简单来说就是对 RocksDB 的 partitioned Index 做了多级索引

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

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.flink.tuning.RocksdbTuning \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar

简单来说，Flink 拓扑中每个节点（Task）间的数据都以阻塞队列的方式传输，下游来
不及消费导致队列被占满后，上游的生产也会被阻塞，最终导致数据源的摄入被阻塞

影响 checkpoint 时长

影响 state 大小

危害

Flink 1.13 优化了反压检测的逻辑（使用基于任务 Mailbox 计时，而不在再于堆栈采 样），并且重新实现了作业图的 UI 展示：Flink 现在在 UI 上通过颜色和数值来展示繁忙 和反压的程度

常用的几个 Metrics

根据指标分析反压

可以进一步分析数据传输

（1）该节点的发送速率跟不上它的产生数据速率。这一般会发生在一条输入多条输出
的 Operator（比如 flatmap）。这种情况，该节点是反压的根源节点，它是从 Source Task 
到 Sink Task 的第一个出现反压的节点

（2）下游的节点接受速率较慢，通过反压机制限制了该节点的发送速率。这种情况，
需要继续排查下游节点，一直找到第一个为 OK 的一般就是根源节点

我们可以通过 Web UI 各个 SubTask 的 Records Sent 和 Record Received 来确认，另外 Checkpoint detail 里不同 SubTask 的 State size 也是一个分析数据倾斜的有用指标。

如果不是数据倾斜，最常见的问题可能是用户代码的执行效率问题（频繁被阻塞或者性 能问题），需要找到瓶颈算子中的哪部分计算逻辑消耗巨大。 最有用的办法就是对 TaskManager 进行 CPU profile，从中我们可以分析到 Task  Thread 是否跑满一个 CPU 核：如果是的话要分析 CPU 主要花费在哪些函数里面；如果 不是的话要看 Task Thread 阻塞在哪里，可能是用户函数本身有些同步的调用，可能是 checkpoint 或者 GC 等系统活动导致的暂时系统暂停

纵向是调用链，从下往上，顶部就是正在执行的函数
横向是样本出现次数，可以理解为执行时长。
看顶层的哪个函数占据的宽度最大。只要有"平顶"（plateaus），就表示该函数可能存在性能问题

TaskManager 的内存以及 GC 问题也可能会导致反压，包括 TaskManager JVM 各 区内存不合理导致的频繁 Full GC 甚至失联。通常建议使用默认的 G1 垃圾回收器

通过 GC 日志分析出单个 Flink Taskmanager 堆总大小、年轻代、老年代分配的内 存空间、Full GC 后老年代剩余大小等

如果发现我们的 Source 端数据读取性能比较低或者 Sink 端写入性能较差，需要检 查第三方组件是否遇到瓶颈，还有就是做维表 join 时的性能问题

1、不能直接用二次聚合来处理

2、使用 LocalKeyBy 的思想

如果 keyBy 之前就存在数据倾斜，上游算子的某些实例可能处理的数据较多，某些实 例可能处理的数据较少，产生该情况可能是因为数据源的数据本身就不均匀，例如由于某些 原因 Kafka 的 topic 中某些 partition 的数据量较大，某些 partition 的数据量较少

这种情况，需要让 Flink 任务强制进行 shuffle。使用 shuffle、rebalance 或 rescale 算子即可将数据均匀分配，从而解决数据倾斜的问题。

两阶段聚合实现思路