【发布时间】:2016-02-22 11:21:54
【问题描述】:
显示窗口开始和结束时间的方式是什么?类似实现用户定义的窗口?
想知道窗口开始的时间并评估输出是
quantity(WindowAll Sum), window_start_time, window_end_time
12, 1:13:21, 1:13:41
6, 1:13:41, 1:15:01
【问题讨论】:
标签: flink-streaming
显示窗口开始和结束时间的方式是什么?类似实现用户定义的窗口?
想知道窗口开始的时间并评估输出是
quantity(WindowAll Sum), window_start_time, window_end_time
12, 1:13:21, 1:13:41
6, 1:13:41, 1:15:01
【问题讨论】:
标签: flink-streaming
找到了答案。 TimeWindow.class 有 getStart() 和 getEnd()
示例用法:
public static class SumAllWindow implements AllWindowFunction<Tuple2<String,Integer>,
Tuple3<Integer, String, String>, TimeWindow> {
private static transient DateTimeFormatter timeFormatter =
DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ss.SS").withLocale(Locale.GERMAN).
withZone(DateTimeZone.forID("Europe/Berlin"));
@Override
public void apply (TimeWindow window, Iterable<Tuple2<String, Integer>> values,
Collector<Tuple3<Integer, String, String>> out) throws Exception {
DateTime startTs = new DateTime(window.getStart(), DateTimeZone.forID("Europe/Berlin"));
DateTime endTs = new DateTime(window.getEnd(), DateTimeZone.forID("Europe/Berlin"));
int sum = 0;
for (Tuple2<String, Integer> value : values) {
sum += value.f1;
}
out.collect(new Tuple3<>(sum, startTs.toString(timeFormatter), endTs.toString(timeFormatter)));
}
}
在main()中
msgStream.timeWindowAll(Time.of(6, TimeUnit.SECONDS)).apply(new SumAllWindow()).print();
【讨论】: