Flink timeWindow获取开始时间

Question

我正在计算一个时间窗口上的计数（求和1），如下所示：

mappedUserTrackingEvent
            .keyBy("videoId", "userId")
            .timeWindow(Time.seconds(30))
            .sum("count")

我还想将窗口开始时间添加为一个关键字段。所以结果会是这样的：

key: videoId=123,userId=234,time=2016-09-16T17:01:30
value: 50

所以本质上是按窗口聚合计数。最终目标是绘制这些窗口的直方图。

如何将窗口的开头添加为键中的字段？然后，在这种情况下，将窗口对齐到00秒或30秒？这可能吗？

Answer 1

您可以使用方法“聚合”而不是“求和”
在“聚合”中，设置第二个参数实现窗口函数或扩展处理窗口函数
我正在使用Flink-1.4.0，建议使用ProcessWindowFunction，例如：

mappedUserTrackingEvent
    .keyBy("videoId", "userId")
    .timeWindow(Time.seconds(30))
    .aggregate(new Count(), new MyProcessWindowFunction();

public static class MyProcessWindowFunction extends ProcessWindowFunction<Integer, Tuple2<Long,  Integer>, Tuple, TimeWindow>
{
    @Override
    public void process(Tuple tuple, Context context, Iterable<Integer> iterable, Collector<Tuple2<Long,  Integer>> collector) throws Exception
    {
        context.currentProcessingTime();
        context.window().getStart();
    }
}

Answer 2

WindowFunction的Application（）方法提供了一个Window对象，如果您使用keyBy（）. timeWindow（），它是一个TimeWindow。TimeWindow对象有两个方法，getStart（）和getEnd（），分别返回窗口开始和结束的时间戳。

目前无法将sum（）聚合与WindowFunction一起使用。您需要执行以下操作：

 mappedUserTrackingEvent
        .keyBy("videoId", "userId")
        .timeWindow(Time.seconds(30))
        .apply(new MySumReduceFunction(), new MyWindowFunction());`

MySumReduceFunction实现了ReduceFunction接口，并通过增量聚合到达窗口的元素来计算总和。MyWindowFunction实现了WindowFunction。它通过Iterable参数接收聚合值，并使用从TimeWindow参数获得的时间戳丰富值。