Flink KeyBy性能优化咨询：如何无需KeyBy使用状态或提升其速度

我正在对一款从Kafka读取数据、经转换后写入另一Kafka主题的Flink应用进行性能基准测试。为避免同一order-id的消息被当作全新订单处理，需保留上下文，因此通过继承RichFlatMapFunction类并使用ValueState来实现该需求。据我理解，需先调用keyBy得到KeyStream才能执行flatMap，代码如下：

env.addSource(source()).keyBy(Order::getId).flatMap(new OrderMapper()).addSink(sink());

问题在于keyBy操作带来了80至200ms的延迟，若移除keyBy并将flatMap替换为map函数，90分位延迟约为1ms。请问是否可以无需使用keyBy就能使用状态/上下文，或有办法提升keyBy的速度？

一、能否绕过keyBy使用状态？ #

不行。Flink的Keyed State（比如你用到的ValueState）是严格绑定在KeyStream之上的，只有通过keyBy将流转换为KeyStream后，才能基于每个key维护独立的状态上下文。这是Flink状态模型的核心设计——状态与key一一对应，没有keyBy就无法实现按order-id隔离状态，也就没法避免同一order-id的消息被重复当作新订单处理。

二、提升keyBy性能的实用方法 #

1. 对齐并行度，避免不必要的shuffle

• 确保Kafka Source的并行度与源Kafka主题的分区数一致，同时让keyBy之后的flatMap、sink算子并行度与Source保持相同。这样Flink会优先使用本地数据转发，减少跨节点的数据传输，大幅降低keyBy带来的shuffle延迟。
• 例如：如果源Kafka主题有12个分区，就把Source、flatMap、sink的并行度都设为12，避免数据在节点间来回拷贝。

2. 优化key的序列化与计算

• 尽量使用基本类型（如Long、String）作为keyBy的key，这类类型的序列化效率远高于自定义复杂类型。如果必须用自定义类型，一定要实现高效的TypeSerializer，避免默认序列化带来的额外开销。
• 可以在Source阶段提前计算order-id的哈希值，后续keyBy直接使用预计算的哈希值，减少keyBy阶段的计算量。

3. 调整Flink运行时参数

• 开启算子链化：确保pipeline.operator-chaining.enabled参数为true（默认开启），让Source、keyBy、flatMap尽可能链在一起运行，减少算子间的数据传输开销。
• 增加网络缓冲区：调大taskmanager.network.numberOfBuffers参数，避免因缓冲区不足导致的数据阻塞，缓解shuffle阶段的延迟。
• 优化检查点配置：如果检查点间隔设置过短，会频繁触发状态快照拖慢性能。适当调大execution.checkpointing.interval，同时开启增量检查点（针对RocksDB状态后端），减少检查点的开销。

4. 选择合适的状态后端

• 如果你的状态量不大，优先使用MemoryStateBackend，它基于内存存储状态，序列化和读写速度远高于磁盘存储的RocksDBStateBackend。
• 若状态量较大必须用RocksDB，开启state.backend.rocksdb.enable.incremental.checkpoint参数启用增量检查点，减少每次检查点的IO开销。

5. 排查其他潜在延迟源

• 检查OrderMapper中的业务逻辑是否存在阻塞操作（如同步IO、复杂计算），这些操作的延迟可能被误判为keyBy的问题。
• 借助Flink Web UI监控各个算子的延迟、吞吐量指标，定位是否真的是keyBy的shuffle环节导致的延迟，还是其他算子的瓶颈。

内容的提问来源于stack exchange，提问作者Abidi