基于事件时间的Flink时间窗口不输出任何内容

Question

在流处理问题中，我们有3个传感器，每个传感器每8毫秒生成一个时间戳样本（传感器的时间是同步的）。所以我想合并每个时间戳的数据（对于3个传感器，我们应该为每个时间戳输出3个合并的样本数据）。此外，我们有一个160毫秒的时间限制，这样每个数据在生成时间戳后最多应该在160毫秒后输出。所以我决定使用Flink事件时间概念和时间窗口。因为时间戳在每个传感器的样本中都是唯一的，所以我们认为它是数据流的关键。

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
SingleOutputStreamOperator<Tuple3<String,Long, JSONObject>> res = aggregatedTuple
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, Long, JSONObject>>(Time.milliseconds(160)) {

            @Override
            public long extractTimestamp(Tuple3<String, Long, JSONObject> element) {
                return element.f1 ;
            }
        }).keyBy(1).timeWindow(Time.milliseconds(160))
                .allowedLateness(Time.milliseconds(2))
                .sideOutputLateData(lateOutputTag)
                .reduce(Do merging samples);

在代码中，我们首先引入流的第二个字段作为事件时间，并设置周期性水印（因为数据是以固定的速率生成的，具有固定的延迟）。之后，我们将事件时间设置为流的关键。我们也想收集后期数据，所以我们使用了side OutputLateData。最后，我们用相同的键减少（合并）数据。问题是在Flink事件时间模式下，定义的窗口不输出任何数据！如果不设置事件时间，它将输出数据，但我想使用事件时间作为Flin窗口的时间。我尝试了多次窗口和水印，但他们没有输出任何东西。

我使用Flink count窗口和自定义超时触发器成功地解决了相同的问题。

更新：传入的数据流格式为以下类型（如果我们有3个传感器）：

sensor_id, timestamp, data
(1, 1531980773390, {})
(2, 1531980773390, {})
(3, 1531980773390, {})
(1, 1531980773398, {})
(2, 1531980773398, {})
(3, 1531980773398, {})

以此类推，每8毫秒一次。

在数据流中保存延迟数据

final OutputTag<Tuple3<String, Long, JSONObject>> lateOutputTag = new OutputTag<Tuple3<String, Long, JSONObject>>("late-data") {
        };
DataStream<Tuple3<String, Long, JSONObject>> lateData = res.getSideOutput(lateOutputTag);

Answer 1

我认为你的问题是你按时间戳键。

您应该有keyBy（0），以便按传感器id分割流。

Answer 2

难道一切都晚了？

默认情况下，周期性水印生成器每200msec调用一次（实时测量，而不是事件时间）。在你的情况下，这可能相当长。使用setAutoWatermark Interval更改此值。您可能还需要考虑使用setBufferTimeout来减少延迟。