Flink中是否存在保留顺序的多有序流合并算子？

Flink没有开箱即用的归并排序式合并有序流的算子，但可以通过自定义算子实现需求，以下是针对不同场景的解决方案：

针对两个有序流的合并 #

使用connect + CoFlatMap算子，基于归并排序的逻辑维护两个输入流的缓存队列，每次输出队列头部的最小元素：

// 假设已定义两个全局有序的数据流stream1和stream2
stream1.connect(stream2)
    .flatMap(new CoFlatMapFunction<Integer, Integer, Integer>() {
        private final Queue<Integer> queue1 = new LinkedList<>();
        private final Queue<Integer> queue2 = new LinkedList<>();

        @Override
        public void flatMap1(Integer val, Collector<Integer> out) {
            queue1.add(val);
            emitMin(out);
        }

        @Override
        public void flatMap2(Integer val, Collector<Integer> out) {
            queue2.add(val);
            emitMin(out);
        }

        private void emitMin(Collector<Integer> out) {
            while (!queue1.isEmpty() && !queue2.isEmpty()) {
                if (queue1.peek() <= queue2.peek()) {
                    out.collect(queue1.poll());
                } else {
                    out.collect(queue2.poll());
                }
            }
        }
    });

针对三个及以上有序流的合并 #

先通过union合并所有流，再用自定义ProcessFunction维护小顶堆（优先级队列），每次输出堆顶的最小元素：

// 假设已定义多个全局有序的数据流stream1、stream2、stream3
DataStream<Integer> unioned = stream1.union(stream2, stream3);

unioned.process(new ProcessFunction<Integer, Integer>() {
    private final PriorityQueue<Integer> minHeap = new PriorityQueue<>();

    @Override
    public void processElement(Integer val, Context ctx, Collector<Integer> out) {
        minHeap.add(val);
        // 因输入流有序，新元素不会比之前的小，直接输出所有小于等于当前元素的堆顶元素
        while (!minHeap.isEmpty() && minHeap.peek() <= val) {
            out.collect(minHeap.poll());
        }
    }
});

关键注意事项 #

• 上述方案仅适用于输入流全局严格有序的场景；如果是按key局部有序的流，需先通过keyBy分区，再针对每个key的子流执行归并逻辑。
• 处理无限流时，需注意内存占用，可结合Watermark机制，当水位线推进时输出所有小于水位线的缓存元素，避免OOM。

内容的提问来源于stack exchange，提问作者akurmustafa