本文属于 Flink 在生产环境的大规模 CPU 优化实战,大并发任务预计节省 30~50% 的 CPU 消耗。下文会详细分析优化相关的实现原理、问题定位以及优化过程。往往在做性能优化时就会发现:当已经定位到性能瓶颈时,很容易想到优化思路去解决或优化。但定位问题的过程其实是最难的,也就是找性能瓶颈的这个过程更具有意义。本文问题定位的过程以及用到的性能分析工具、命令可能对于广大的技术同学更有收益。
0、 结论
Flink 大并发任务(超过 500 并发)在使用 keyBy 或者 rebalance 的情况下,将 bufferTimeout 设置为 1s 可以节省 30~50% 的 CPU 消耗。中等并发任务也会有不少收益。
1、 Flink 如何权衡吞吐与延迟的关系
TM 是真正处理数据的进程,如下图所示,上游 Task 2 的输出做为下游 Task 1 的输入,必然会产生 shuffle,也就是网络间数据交换。
如果 Task2 每处理一条数据就通过网络发送一条数据给 Task1,显然数据传输效率比较低。为了提高吞吐,Flink 设计了 NetworkBuffer 来实现攒批。即:Task2 的每个 subtask 内部都会缓存一批数据再发送给下游 Task1。
1.1 网络传输 buffer 实现原理
假设一个任务有两个 Task,分别是 TaskA 和 TaskB,TaskA 的并行度为 2,TaskB 的并行度为 3,且 TaskA 和 TaskB 之间的连接方式是 keyBy。如下图所以,总共有 2 * 3 = 6 个连接。
详细的数据传输如下图所示。上游每个 subtask 中会有 3 个 resultSubPartition,连接下游算子的 3 个 subtask。下游每个 subtask 中会有 2 个 InputChannel,连接上游算子的 2 个 subtask。一个个彩色的小圆圈就是一个个 buffer,buffer 生成以后,就可以发送到下游 TaskB。
Note:Subtask 中 buffer 生成好以后,会被 Netty 消费,由 Netty Server 负责将数据发送给下游 Subtask B。SubtaskB 的 Netty Client 将接收到的 buffer 写到 InputChannel 中,下游的 Subtask 就可以开始反序列化处理数据了。具体参考官网图示:
1.2 生成 buffer 的规则
Netty Server 会消费 SubtaskA resultSubPartition 产生的 buffer 数据,那 NettyServer 什么时候能消费到 buffer 呢?
正在写入的 buffer 肯定不能被 Netty 消费到,只有写完的 buffer 才能被消费到。在 Flink 中有三种情况会认为 buffer 写完了,可以被 Netty Server 消费:
buffer 写满了
buffer 超时了
遇到特殊的 Event,例如:Checkpoint barrier
默认 buffer 大小是 32KB,如果 32KB 写满就认为当前 buffer 写完了,可以将当前 buffer 发送到下游 Task,也就是条件1。
如果低峰期数据量比较小,1 分钟也没有将 32KB 的 buffer 写满,那么数据将一直缓存在 buffer 中,从而导致大量的延迟。为了解决数据延迟的问题,Flink 增加了一个时间策略,默认是 100ms。如果 100ms buffer 还没写满,为了保障数据的实时性,也认为 buffer 写完了,可以将当前 buffer 发送到下游 Task,也就是条件 2。
遇到一些特殊的事件,例如遇到 Checkpoint barrier,为了加快 Checkpoint 效率,会直接认为 buffer 写完了,可以将当前 buffer 发送到下游 Task,也就是条件 3。
当然条件 2 的 100ms 是可以调节的,Flink 1.10 及以后的版本直接通过配置参数execution.buffer-timeout: 100ms
可以设置,Flink 1.10 之前通过代码env.setBufferTimeout(100);
可以设置。
1.3 小结
Flink 通过 buffer-timeout 参数对吞吐和延迟做权衡,当设置为 0 时表示没有 timeout 策略,即:每条数据来了都认为 buffer 满了,将这一条数据单独发送给下游。保障了实时性,但吞吐可能会下降。要想吞吐不下降,就需要消耗更多的资源。
100ms 是 Flink 权衡过的 timeout 默认值,既能保证吞吐,又能保障延迟控制在 100ms 以内。
2、 大并发任务为什么会消耗更多的 CPU?
2.1 大并发的 resultSubPartition 图
假设 TaskA 有 1000 个并发,下游 TaskB 也有1000 个并发,意味着 SubtaskA0 总共有 1000 个 resultSubPartition,分别对应 SubtaskB0、B1 ... B999。如下所示是 SubtaskA0 的 resultSubPartition 图,其他的 SubtaskA1、A2 ... A999 也是同样的图示,图中可以看到 SubtaskA0 中会有 1000 个 resultSubPartition。
2.2 大并发情况,每个 buffer 中会缓存多少条数据?
buffer 是为了缓存一批数据,批量发送,提高效率。如果是大并发情况,到底每个 buffer 会缓存多少条数据做为一个批次呢?
默认 bufferTimeout = 100ms 意味着 100ms 必须将 buffer 中的数据发送到下游 Task。假设 SubtaskA0 每秒处理 1 万条数据,则 100ms 平均处理 10000 / 10 = 1000 条数据。这 1000 条数据要发送给 1000 个 resultSubPartition,所以平均一个 resultSubPartition 中只会有 1 条数据,每个 resultSubPartition 有自己独立的 buffer,也就是说每个 buffer 其实只缓存了 1 条数据。跟没有 buffer 没什么区别。。。
如果 TaskB 的并发为 2000,则 SubtaskA0 内部会有 2000 个 resultSubPartition。100ms 产生 1000 条数据,则只有 1000 个 resultSubPartition 中有数据,仍然是 100ms 发送 1000个 buffer 给下游 TaskB,每个 buffer 中只有 1 条数据。
2.3 大并发任务消耗 CPU 小结
生产环境很多任务的并发远大于 1000,所以造成很多 buffer 仅仅只缓存 1 条数据就被 timeout 策略触发发送给下游 Task。每条数据做为 1 个 buffer,每秒处理 1 万条数据,则后台线程每秒需要 flush 1 万个 buffer 到 NettyServer,从而大量消耗 CPU。
不仅是发送方效率降低,下游的 Subtask B 接受数据的效率也会降低。每秒接受 1 万的 buffer,每个 buffer 里 1 条数据。大量的小 buffer,大量的读取小数据,消耗大量的 CPU 资源。
2.4 小并发任务会存在 buffer 攒批效果不好的问题吗?
假设 TaskA 和 Task B 的并发是 50,SubtaskA 对应 50 个 resultSubPartition。每秒产生 1 万条数据,则 100ms 需要会处理 1000 条数据,平均每个 resultSubPartition 能攒 1000 / 50 = 20 条数据。相对而言攒批效果还算可以。
如果单条消息比较大,可能 20 条数据早就将 buffer 占满了,buffer 如果占满会直接发送到下游 Task,无需等待 timeout 超时才发送。
已经举了多个案例,大家可以使用上述的计算方式计算一下,生产环境自己开发的任务是否存在:因为攒批效果不好导致 CPU 消耗较多的任务?
3、 优化思路、测试结果及其优化收益
3.1 优化思路
经过上面的理论分析,很容易想到的优化思路就是:调大 bufferTimeout 参数。
TaskA 和 Task B 的并发仍然是 1000,将 bufferTimeout 调大到 1s。SubtaskA 对应 1000 个 resultSubPartition,每秒产生 1万条数据,则每个 resultSubPartition 平均分配 10 条数据,即:每个 buffer 平均缓存了 10 条数据,相比之前 1 条来讲,buffer 攒批的效果明显变好。
3.2 测试结果
测试条件:业务逻辑简单,keyBy 后不做任何业务处理,并发 1200。
因为没有业务处理,所以从业务角度来讲,CPU 应该主要消耗在序列化和反序列化 protobuf。从引擎角度来讲,TM 之间数据 flush、传输需要消耗不少的 CPU。
基于上述条件,仅调节 bufferTimeout 参数,观察单个 TM 的平均 CPU 消耗。
bufferTimeout = 100ms 时,TM 平均 CPU 使用 0.59 core
bufferTimeout = 1s 时,TM 平均 CPU 使用 0.39 core,CPU 节省了 33%
bufferTimeout = 10s 时,TM 平均 CPU 使用 0.33 core,CPU 节省了 44%
bufferTimeout 无限扩大并不能一直提高性能,因为 3 秒可能 buffer 就满了,此时设置 3 s 和 10s 并没有任何区别。
不同业务方的多个同学反馈:大多数业务场景都可以接受 1s 延迟,所以生产环境将大并发任务的 bufferTimeout 设置为 1s 是个不错的选择。既能节省不少的 CPU 消耗,又能保证延迟在 1s 以内。
3.3 生产环境任务的实际收益
测试结果毕竟没有生产环境更具有说服力。与多个业务线的同学沟通后,调整了 3 个大任务,CPU 最少的节省了 30%,最多的节省了 50%。
4、 问题分析定位的过程
上述部分都是在讲原理、优化过程以及最后的测试结果和优化后的收益。往往在做性能优化时就会发现:当已经定位到性能瓶颈时,很容易想到优化思路去解决或优化。但定位问题的过程其实是最难的,也就是找性能瓶颈的这个过程更具有意义。
下面主要讲述:如何发现调节 bufferTimeout 参数会节省很多的 CPU?一些性能优化的方法论。
其实很多同学都知道 bufferTimeout 参数调大可能会减少资源消耗,但却没想到大并发任务调大 bufferTimeout 后会有如此大的资源收益。
4.1 火焰图
业务方反馈有个 数千并发的 Flink 任务高峰期单个 TM CPU 使用 150% 以上,需要架构组同学辅助做优化。
划重点:CPU 使用比较高,要对 CPU 使用做优化,首先需要知道 CPU 大部分时间在做什么工作?假设 CPU 有 1% 的时间在执行 A,30% 的时间在执行 B,69% 的时间在执行 C。那么应该花更多的精力去优化 C。
至于那 1% 的 A 可能也有优化空间,优化后可能也有收益,但是由于占得比例特别低,就算优化了,收益也比较小,没必要耗费大量的精力去做 A(性能优化也要考虑 ROI)。
那有没有工具能分析进程的 CPU 在频繁执行哪些代码吗?有,那就是《火焰图》。
火焰图能呈现出应用程序在执行哪些方法,每个方法执行的时长占了整个 CPU 的消耗的百分比。由于篇幅原因,本文不会详细介绍火焰图,关于火焰图的生成以及如何读懂火焰图,大家可以自行学习。
下图是一张火焰图,可以看到 netty 相关线程的方法占了 29.38% 的 CPU。
使用火焰图对进程分析以后,发现有几大块 CPU 消耗比较大:
netty 相关
kafka 序列化压缩消息
protobuf 序列化反序列化
此时也没办法确定 netty 这块可以优化,以为大数据量下数据传输正常就需要这么多 CPU。至于后两者属于业务角度的 CPU 消耗,暂且忽略。
4.2 创建测试任务用于性能分析
线上任务没办法用于测试,所以需要创建一个新的副本任务用于测试。如果消费全量数据测试,则需要相同的资源量(几千的 TM),显然没有那么多的资源。
幸运的是我们团队的刚哥开发了一个 Feature:通过参数控制只消费 Kafka 部分的 partition。假设 kafka 有 2000 个 partition,可以通过参数指定只消费十分之一的 partition,也就是 200 个 partition。这样就只需要十分之一的资源用于测试即可。感谢刚哥开发的超级实用的 Feature。
有些任务消费十分之一的 partition,资源并不能减少为原来的十分之一,例如计算 uv 的场景,如果仅仅消费十分之一的数据,uv 应该不能减少原来的十分之一。有些任务消费十分之一的 partition,资源可以减少为原来的十分之一,例如计算 pv 的场景。
幸运的是:当前要优化的任务虽然不是简单的 pv 场景,但如果消费十分之一的 partition,资源理论来讲应该是原来的十分之一左右。
所以就创建了一个测试任务,相比线上任务做了以下三个改动:
Sink 替换成测试的 topic
Source 仅消费十分之一的 partition
并行度改为生产环境的十分之一
4.3 分析性能
理论来讲,测试任务的 TM 平均 CPU 使用率与生产环境基本一样。因为数据量少了十倍,资源少了十倍。
但是发现测试任务的 CPU 使用率要比生产环境低 30%,不可思议。继续使用火焰图分析,对比测试任务和线上任务的 CPU 使用到底差在哪里?发现主要区别在于 netty 网络相关的 CPU 消耗:
线上任务的 netty 占整个 CPU 使用率的 30%
测试任务的 netty 占整个 CPU 使用率的 6%
两个任务的区别主要在于并发数,测试任务的 TM 数量是线上任务的十分之一。
所以初步怀疑:大并发情况,单个 TM 处理的数据量虽然不变,但网络间数据传输消耗的 CPU 会变多。
根据之前的计算公式:如果 SubtaskA 处理的数据量不变,下游并发数越多对应的 resultSubPartition 就会越多,导致每个 resultSubPartition 中的 buffer 攒批效果不好。反之下游并发数越少对应的 resultSubPartition 就会越少,则每个 resultSubPartition 中的 buffer 攒批变好,从而提高了性能。
于是猜想:任务的并发数不变,如果调大 bufferTimeout 时间,攒批效果也会变好,是不是性能也会有提升呢?
4.4 验证猜想
于是就有了本文 3.2 部分的测试结果。写了一个业务逻辑特别简单的任务,仅调节 bufferTimeout 参数,观察单个 TM 的平均 CPU 消耗。
bufferTimeout = 100ms 时,TM 平均 CPU 使用 0.59 core
bufferTimeout = 1s 时,TM 平均 CPU 使用 0.39 core,CPU 节省了 33%
bufferTimeout = 10s 时,TM 平均 CPU 使用 0.33 core,CPU 节省了 44%
于是就确信了:优化攒批效果会节省很多的 CPU 消耗。这也就是整个性能分析优化的过程。
5、 总结
Flink 依赖 NetworkBuffer 的 timeout 参数对吞吐与延迟做权衡。大并发任务如果使用 keyBy 或者 rebalance,则 buffer 的攒批效果会下降,每个 buffer 里存放一条数据,导致整体资源消耗变大。可以调大 bufferTimeout 将 buffer 中数据变多,提高攒批效果,从而优化资源消耗。
做性能分析和性能优化要合理使用科学的性能分析工具,合理的工具能实现事半功倍。当有很多优化思路时,性能分析工具可以在优化之前分析出每个优化项预期有多大收益。优先去做那些 ROI 比较高的优化。
如果觉得《性能优化:一个 Flink 参数节省了 50% 的 CPU 消耗》对你有帮助,请点赞、收藏,并留下你的观点哦!