0

我有以下 Flink 管道，它简单地计算窗口中的元素并在单独的流上报告后期元素

OutputTag<Tuple3<Long, String, Double>> lateItems= new OutputTag<Tuple3<Long, String, Double>>("Late Items"){};
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

env.setParallelism(1);
env.getConfig().setAutoWatermarkInterval(-1);
DataStream<Tuple3<Long,String,Double>> stream = env.addSource(new  YetAnotherSource(fileName));
DataStream<Tuple3<Long,String,Double>> lateStream;
AllWindowedStream<Tuple3<Long, String, Double>, TimeWindow> tuple3TimeWindowAllWindowedStream = stream.windowAll(SlidingEventTimeWindows.of(Time.milliseconds(100), Time.milliseconds(10)));
tuple3TimeWindowAllWindowedStream.sideOutputLateData(lateItems);
lateStream = streamOfResults.getSideOutput(lateItems);
            lateStream.countWindowAll(1).apply(new CounterFunction22()).writeAsText("FlinkSlidingTimeWindowLateItemsResult.txt",FileSystem.WriteMode.OVERWRITE);
streamOfResults.writeAsText("FlinkSlidingTimeWindowOutputFor" + fileName + ".txt", FileSystem.WriteMode.OVERWRITE);

当我将以下数据作为输入传递时

1383451269002,A,22
1383451269006,A,18
1383451269007,A,18
*1383451269010,W,0
1383451269008,A,18
1383451269027,A,20
1383451269028,A,19
1383451269033,A,17
1383451269033,A,17
1383451269030,A,17
*1383451269038,W,0
1383451269008,A,17

带 * 的元素是水印。

我得到了预期的结果，第一个窗口包含三个元素。那是因为第五行和最后一行的元素对于该窗口被认为是迟到的。

(1383451268910,1383451269010,3)

然而，在侧面输出没有产生任何东西。

但是，当我使用会话窗口时，会在侧面输出中生成迟到的项目。

任何想法为什么没有为滑动时间窗口生成任何东西？

4

0 回答 0