Flink缺少窗户处理器（事件时间窗口）和Kafka源

Question

我们有一个流媒体作业，具有20个单独的管道，每个管道都有一个/许多Kafka主题来源，并且一些管道带有窗户处理器，而其他管道则是未窗口的处理器。

当作业下降时，我们注意到窗户处理器管道的数据丢失，并且需要一些时间才能恢复/何时需要重新启动作业。

• 我已经为所有操作员设置了UID，并且可以在日志中看到，从kafka消费者操作员的保存点还原偏移量

• 我们使用boundedOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOutOffordersTimestAmpAmpExtractor来根据事件时间分配水印。

  。

public class KafkaEventTimestampExtractor extends BoundedOutOfOrdernessTimestampExtractor<Event> implements Serializable{

    public KafkaEventTimestampExtractor(Time maxOutOfOrderness) {
        super(maxOutOfOrderness);
    }

    @Override
    public long extractTimestamp(Event element) {
        try {
            log.info("event to be processed, event:{}", new ObjectMapper().writeValueAsString(element));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        Long ts = null;
        ts = Double.valueOf(Double.parseDouble(element.getTs())).longValue();
        ts = ts.toString().length() < 13 ? ts * 1000 : ts;
        return ts;
    }
}

管道配置看起来像这样。

• 非窗户的

SourceUtil
  .getEventDataStream(env, kafkaSourceSet)
  .process(new S3EventProcessor()).uid(“…..**)
  .addSink();

• 窗口

SourceUtil
  .getEventDataStream(env, kafkaSourceSet)
  .assignTimestampsAndWatermarks(
    new KafkaEventTimestampExtractor(Time.seconds(4)))
  .windowAll(TumblingEventTimeWindows.of(
    Time.milliseconds(kafkaSourceSet.bufferWindowSize))
  .process(new S3EventProcessor()).uid(“…..**)
  .addSink();

• 说，工作是30分钟的下降，在这种情况下，我们不使用窗口处理器不会错过任何数据，但是在那30分钟内从窗户处理器中丢失了Paritial Data。

  。

• 当我们增加timeWinows的序列外事件延迟时，即我们将其从4sec增加到30分钟，如果应用程序在30分钟内，则不会错过这些事件。我们远不及解决方案。对于我们来说，超过1分钟的延迟也不可见，也会有太多的现场窗户对我们来说意味着巨大的下属变化。

Answer 1

我可以想象的唯一情况可能会解释这一点是事件时间戳是否受到停机的影响。然后，30分钟的停电将在时间戳中导致30分钟的间隙，并且随着外部摄入，4秒的订购策略将产生一些后期事件，这些事件将在窗户上删除。

Answer 2

这是由于我的管道中的错误而发生的，而不是使用FlinkkafkaconSumer的时间戳分配程序，而是将其添加到FlinkkafkaConsumer生成的数据流中。

这一更改已解决了我的问题，以自动恢复，但是如果手动重新启动管道的任何更改，则在作业停止时，最后一个窗口仍缺少某些数据。

注意： - 我们正在使用检查点进行手动恢复。
根据文档，检查点是工作失败时自动恢复的理想选择。

如果我们需要创建一个保存点，以防我们需要对管道进行一些更改并手动重新启动它，或者我们可以通过检查点进行完全恢复，那么对此有所帮助。

我们唯一关心的是使用SavePoint，是对可能发生的相同事件的重新处理，这对我们而言并不理想。