在翻转窗口后执行闪烁下沉

Question

来源：Kinesis数据流

水槽：弹性搜索

对于使用AWS服务的两者。

此外，在AWS Kinesis数据分析应用程序上运行我的Flink作业

我面临着flink窗口功能的问题。我的工作是这样的

DataStream<TrackingData> input = ...; // input from kinesis stream
input.keyBy(e -> e.getArea())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .reduce(new MyReduceFunction(), new MyProcessWindowFunction())
                .addSink(<elasticsearch sink>);

 private static class MyReduceFunction implements ReduceFunction<TrackingData> {
        @Override
        public TrackingData reduce(TrackingData trackingData, TrackingData t1) throws Exception {
            trackingData.setVideoDuration(trackingData.getVideoDuration() + t1.getVideoDuration());
            return trackingData;
        }
    }

private static class MyProcessWindowFunction extends ProcessWindowFunction<TrackingData, TrackingData, String, TimeWindow> {
        public void process(String key,
                            Context context,
                            Iterable<TrackingData> in,
                            Collector<TrackingData> out) {

            TrackingData trackingIn = in.iterator().next();

            Long videoDuration =0l;
            for (TrackingData t: in) {
                videoDuration += t.getVideoDuration();
            }
            trackingIn.setVideoDuration(videoDuration);
            out.collect(trackingIn);
        }
    }

示例事件：

{"area":"sessions","userId":4450,"date":"2021-12-03T11:00:00","videoDuration":5} 

我在这里所做的是从动觉流中，我得到了大量的这些事件，我想对每10秒窗口的视频持续时间求和，然后我想将单个事件存储到elasticsearch中。

在动觉中，每秒可以有10000个事件。我不想在elasticsearch中存储所有10000个事件，我只想每10秒存储一个事件。

问题是，当我向此作业发送事件时，它会快速处理此事件并直接进入elasticsearch，但我希望实现：直到每10秒，我希望事件视频持续时间时间递增，10秒后，elasticsearch中只存储一个事件。

我如何才能实现这一点？

Answer 1

我想你把这个问题误诊了。

您编写的代码将从每10秒长的窗口中为窗口中包含事件的每个不同键生成一个事件<代码>MyProcessWindowFunction没有任何效果：由于窗口结果已预先聚合，每个Iterable将只包含一个事件。

我相信你想这样做：

input.keyBy(e -> e.getArea())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .reduce(new MyReduceFunction())
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .reduce(new MyReduceFunction())
                .addSink(<elasticsearch sink>);

你也可以只做

input.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .reduce(new MyReduceFunction())
                .addSink(<elasticsearch sink>);

但是第一个版本会更快，因为它将能够在计算windowAll中的全局总和之前并行计算每个键窗口的结果。

FWIW，表/SQL API通常更适合这种类型的应用程序，并且应该生成比这两种应用程序都更优化的管道。