Flink DataStream排序程序不输出

Question

我用Flink编写了一个小测试用例代码来对数据流进行排序。代码如下：

public enum StreamSortTest {
    ;
    public static class MyProcessWindowFunction extends ProcessWindowFunction<Long,Long,Integer, TimeWindow> {
        @Override
        public void process(Integer key, Context ctx, Iterable<Long> input, Collector<Long> out) {
            List<Long> sortedList = new ArrayList<>();
            for(Long i: input){
                sortedList.add(i);
            }
            Collections.sort(sortedList);
            sortedList.forEach(l -> out.collect(l));
        }
    }

    public static void main(final String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env.getConfig().setExecutionMode(ExecutionMode.PIPELINED);

        DataStream<Long> probeSource = env.fromSequence(1, 500).setParallelism(2);

        // range partition the stream into two parts based on data value
        DataStream<Long> sortOutput =
                probeSource
                        .keyBy(x->{
                            if(x<250){
                                return 1;
                            } else {
                                return 2;
                            }
                        })
                        .window(TumblingProcessingTimeWindows.of(Time.seconds(20)))
                        .process(new MyProcessWindowFunction())
                        ;

        sortOutput.print();
        System.out.println(env.getExecutionPlan());
        env.executeAsync();
    }
}

然而，代码只输出执行计划和其他几行。但它不会输出实际排序的数字。我做错了什么？

Answer 1

我能看到的主要问题是，您使用的是基于处理时间的窗口，输入数据非常短，处理时间肯定会少于20秒。而Flink能够检测到输入的结束（如果是来自文件或序列的流，就像您的情况一样），并生成Long。最大水印，它将关闭所有打开的基于事件时间的窗口，并触发所有基于事件时间的计时器。它对基于处理时间的计算没有同样的作用，因此在您的情况下，您需要断言Flink实际上会工作足够长的时间，以便关闭窗口或引用自定义触发器/不同的时间特征。

还有一件事我不确定，因为我从来没有用过它那么多，那就是您是否应该使用刽子手同步进行本地执行，因为这基本上是针对您不想等待作业结果的情况根据这里的文档。