用于非连续数据的Flink和Kinesis流应用程序

Question

我们已经构建了一个Flink应用程序来处理来自动觉流的数据。应用程序的执行流程包含基于注册类型过滤数据、基于事件时间戳分配水印的基本操作，以及应用于5分钟数据窗口的映射、处理和聚合功能，如下所示：

    final SingleOutputStreamOperator<Object> inputStream = env.addSource(consumer)
            .setParallelism(..)
            .filter(..)
            .assignTimestampsAndWatermarks(..);

    // Processing flow
    inputStream
            .map(..)
            .keyBy(..)
            .window(..)
            .sideOutputLateData(outputTag)
            .aggregate(aggregateFunction, processWindowFunction);

    // store processed data to external storage
    AsyncDataStream.unorderedWait(...);

我的水印分配程序的参考代码：

    @Override
public void onEvent(@NonNull final MetricSegment metricSegment,
                    final long eventTimestamp,
                    @NonNull final WatermarkOutput watermarkOutput) {
    if (eventTimestamp > eventMaxTimestamp) {
        currentMaxTimestamp = Instant.now().toEpochMilli();
    }
    eventMaxTimestamp = Math.max(eventMaxTimestamp, eventTimestamp);
}

@Override
public void onPeriodicEmit(@NonNull final WatermarkOutput watermarkOutput) {
    final Instant maxEventTimestamp = Instant.ofEpochMilli(eventMaxTimestamp);
    final Duration timeElaspsed = Duration.between(Instant.ofEpochMilli(lastCurrentTimestamp), Instant.now());
    if (timeElaspsed.getSeconds() >= emitWatermarkIntervalSec) {
        final long watermarkTimestamp = maxEventTimestamp.plus(1, ChronoUnit.MINUTES).toEpochMilli();
        watermarkOutput.emitWatermark(new Watermark(watermarkTimestamp));
    }
}

现在，这个应用程序的性能很好（在几秒钟内的延迟方面），早就有了。然而，最近在上游系统post中发生了变化，其中Kinesis流中的数据以突发方式发布到流中（每天仅2-3小时）。在这一变化之后，我们看到我们的应用程序的延迟出现了巨大的峰值（使用flinkgauge方法测量，方法是在first filter方法中记录开始时间，然后在Async方法中通过计算从开始时间映射到该点的时间戳中的差异来发出度量）。想知道对于突发流量/非连续数据流，将Flink应用程序与Kinesis流一起使用是否存在任何问题？




Answer 1

由于输入流现在长时间处于空闲状态，这可能会造成水印被挂起的情况。如果是这种情况，那么我预计延迟会有很大的变化，因为它（可能）只是每个突发的最终窗口，其结果会延迟到下一个突发的到来。