flink-streaming - ウィンドウの開始時間と終了時間

Question

ウィンドウの開始時刻と終了時刻を表示するにはどうすればよいでしょうか? ユーザー定義のウィンドウを実装するようなものですか？

出力が

quantity(WindowAll Sum), window_start_time, window_end_time
12, 1:13:21, 1:13:41
6, 1:13:41, 1:15:01

Answer 1

答えを見つけました。TimeWindow.classgetStart() と 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)));
    }
}

メイン()で

msgStream.timeWindowAll(Time.of(6, TimeUnit.SECONDS)).apply(new SumAllWindow()).print();