基于事件时间输出的Flink时间窗口

在流处理问题中,我们有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);

在代码中,我们首先引入流的第二个字段作为事件时间,并设置一个周期性水印(因为数据是以固定的速率和固定的延迟生成的)。之后,我们将事件时间设置为流的键。我们也希望收集最新的数据,所以我们使用 sideOutputLateData . 最后,我们使用相同的键来减少(合并)数据。问题是在Flink事件时间模式下,定义的窗口不输出任何数据!如果不设置事件时间,它将输出数据,但我想使用事件时间作为燧石窗口的时间。我多次尝试了窗口和水印,但没有输出任何内容。

我使用Flink计数窗口和定制的超时触发器成功地解决了同样的问题。

更新: 传入的数据流格式为类型(如果我们有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);