如何协调Flink中流的速度？解决algorithm stream输出因watermark过期被丢弃问题

Hey there! 我之前在做Flink流处理的时候也碰到过几乎一模一样的问题——快流速的原始日志流把水印推得太靠前，导致带窗口的算法流结果还没产出就被当成迟到数据丢弃了。结合实际踩过的坑，给你几个可行的技术方案：

这是最直接的方向，核心是让水印的推进速度匹配算法流的处理能力：

• 设置允许的延迟时间：给窗口算子加上allowedLateness(Time.minutes(5))（时间根据你的实际延迟情况调整），这样就算窗口因为水印触发关闭后，迟到的算法结果仍然能被纳入窗口计算，不会直接丢弃。
• 优化水印生成的乱序容忍度：如果原始日志流的水印是基于自身事件时间生成的，可以增大乱序容忍的时间窗口，比如把WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10))改成Duration.ofMinutes(5)，放慢水印的推进速度。
• 基于关联流的水印对齐：如果两个流需要做关联计算，可以使用WatermarkStrategy.withTimestampAssigner自定义水印生成逻辑，取两个流中事件时间较小的那个作为全局水印，避免其中一个流的水印过快推进。

算法流处理慢才是根源，提升它的吞吐量能从根本上解决延迟问题：

• 调整窗口参数：如果业务允许，把大窗口拆成更小的窗口（比如1小时窗口改成15分钟窗口），减少每个窗口需要处理的数据量；或者用滑动窗口代替滚动窗口，让结果产出更频繁。
• 提升算子并行度：给算法流的窗口算子设置更高的并行度，比如yourWindowOperator.setParallelism(8)，让更多的计算资源来处理窗口数据，加快计算速度。
• 优化状态后端：如果窗口状态较大，把默认的MemoryStateBackend换成RocksDBStateBackend，并开启增量Checkpoint，这样能减少内存压力，同时缩短Checkpoint的耗时，避免因为Checkpoint阻塞导致的处理延迟。
• 开启算子链：确保Flink的算子链优化是开启的（默认开启，如果你手动关闭了可以重新打开），减少算子间的数据传输开销，提升整体处理效率。

如果原始日志流的流速确实远超算法流的处理能力，可以从流控和兜底两个角度入手：

• 对原始日志流限流：使用Flink的RateLimiter算子或者自定义一个限流算子，控制原始日志流的输入速度，让它和算法流的处理能力匹配。不过这个方案要权衡业务对原始日志处理延迟的接受度。
• 使用侧输出流收集迟到数据：在窗口算子中通过sideOutputLateData(new OutputTag<>("late-algorithm-results", TypeInformation.of(YourResultType.class)))把被水印丢弃的迟到结果收集到侧输出流，后续可以单独对这些数据进行补处理，或者和原始日志流做异步关联。

如果业务对时间的准确性要求不是特别严格，可以考虑把算法流的窗口从事件时间窗口改成处理时间窗口：
比如把原来的TumblingEventTimeWindows.of(Time.hours(1))换成TumblingProcessingTimeWindows.of(Time.hours(1))，这样窗口的触发基于算子的处理时间，不会受水印的影响，从根本上避免因为水印推进过快导致的结果丢弃问题。不过这个方案要注意处理时间窗口可能带来的时间偏差问题。

如果算法流的延迟是因为Checkpoint超时或者频繁重启导致的，可以调整相关配置：

• 增大Checkpoint的超时时间：env.getCheckpointConfig().setCheckpointTimeout(Time.minutes(10).toMillis())，避免因为Checkpoint超时导致算子重启。
• 调整Checkpoint间隔：如果当前Checkpoint太频繁，适当增大间隔（比如从1分钟改成5分钟），减少Checkpoint对正常处理的影响。
• 使用固定延迟重启策略：env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.minutes(1)))，避免因为频繁重启导致的处理中断。

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