利用并行性生成有序窗口聚合（例如，前10个查询）

Question

我正在尝试利用并行性来加速一个前10位的窗口操作。我的应用程序由具有时间戳和键和（即tuple2 )的事件组成，我的目标是为30分钟的滚动窗口生成前10个最频繁的键（使用事件时间）。为此，我的查询由一个入口、一个窗口和一个聚合阶段组成。换句话说，我的代码将需要执行类似以下内容的操作：

DataStream<Tuple3<Long, String, Integer>> s = env
    .readTextFile("data.csv")
    .map(new MapFunction<String, Tuple3<Long, String, Integer>>() {
      @Override
      public Tuple3<Long, String, Integer> map(String s) throws Exception {
        String[] tokens = s.split(",");
        return new Tuple3<Long, String, Integer>(Long.parseLong(tokens[0]),
            tokens[1], 1);
      }})
    .assignTimestampsAndWatermarks(
        new AscendingTimestampExtractor<Tuple3<Long, String, Integer>>() {
          @Override
          public long extractAscendingTimestamp(Tuple3<Long, String, Integer> t) {
            return t.f0;
          }}).setParallelism(1);

以上是从一个CSV文件解析数据并分配事件时间（即入口）的代码。我将并行性设置为1的原因是，我需要事件按顺序显示，以便将它们分配给Windows。

接下来是棘手的部分，我试图在生成正确的（有序的）窗口结果的同时加快执行速度。

朴素（串行）执行

DataStream<Tuple2<Long, String>> windowedTopTen = s
        .windowAll(TumblingEventTimeWindows.of(Time.minutes(30)))
        .apply(new SerialAggregation()).setParallelism(1);

DataStream<Tuple2<Long, String>> windowedTopTen = s
    .keyBy(1)
    .window(TumblingEventTimeWindows.of(Time.minutes(30)))
    .apply(new PartialAggregation()).setParallelism(N);

其中partialaggregation()将为不同的时间戳生成部分结果（不相交的键集）。换句话说，我的理解是，对于相同的时间戳t1，我将用partial_result_1到partial_result_n，其中n是我设置的并行性。我的目标是聚合特定时间戳（如t1)的所有部分结果，但我不知道如何做到这一点。此外，当我能够将部分结果与匹配的时间戳组合在一起时，我将如何生成一个数据流，其元组是基于时间戳排序的（就像朴素解决方案产生的结果一样）。

问题

1. 如何完成并行（更快）方法以产生所需结果并将部分结果与匹配的时间戳结合？
2. 在我将每个时间戳的部分结果合并后，是否有方法生成一个数据流，其中结果按时间戳的顺序显示？

Answer 1

首先，如果将Tuple2替换为Tuple3，其中字符串是单个键，整数是计数器，那么将会更容易将部分前10名结果组合为整体前10名。

然后，您可以使用windowAll和一个聚合窗口函数，在窗口的第二层中添加前10个键（总的）及其计数。