如何对无水印闪烁的联合数据流进行排序

Question

flink流有多个数据流，然后我将这些数据流与org合并。阿帕奇。Flink。流式处理。api。数据流。DataStream#联合方法。然后，我得到了问题，数据流是无序的，我不能设置窗口来排序数据流中的数据。

在Apache Flink中对流的联合进行排序以识别用户会话

我得到了答案，但是com。利亚姆。学Flink。实例协会UnionStreamDemo。从未调用SortFunction#onTimer。

环境信息：flink版本1.7.0

一般来说，我希望对没有水印的联合数据流进行排序。

Answer 1

您需要水印，以便排序函数知道何时可以安全地发出已排序的元素。没有水印，您会从流B中获得一条日期早于流A的前N条记录的记录，对吗？

但添加水印很容易，特别是如果您知道任何一个流的“事件时间”都在严格增加。下面是我写的一些代码，它将DavidAnderson在回答上面提到的另一个SO问题时发布的内容扩展到了其他SO问题，希望这能帮助您开始。

--肯

package com.scaleunlimited.flinksnippets;

import java.util.PriorityQueue;
import java.util.Random;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.util.Collector;
import org.junit.Test;

public class MergeAndSortStreamsTest {

    @Test
    public void testMergeAndSort() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(2);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<Event> streamA = env.addSource(new EventSource("A"))
                .assignTimestampsAndWatermarks(new EventTSWAssigner());
        DataStream<Event> streamB = env.addSource(new EventSource("B"))
                .assignTimestampsAndWatermarks(new EventTSWAssigner());

        streamA.union(streamB)
        .keyBy(r -> r.getKey())
        .process(new SortByTimestampFunction())
        .print();

        env.execute();
    }

    private static class Event implements Comparable<Event> {
        private String _label;
        private long _timestamp;

        public Event(String label, long timestamp) {
            _label = label;
            _timestamp = timestamp;
        }

        public String getLabel() {
            return _label;
        }

        public void setLabel(String label) {
            _label = label;
        }

        public String getKey() {
            return "1";
        }

        public long getTimestamp() {
            return _timestamp;
        }

        public void setTimestamp(long timestamp) {
            _timestamp = timestamp;
        }

        @Override
        public String toString() {
            return String.format("%s @ %d", _label, _timestamp);
        }

        @Override
        public int compareTo(Event o) {
            return Long.compare(_timestamp, o._timestamp);
        }
    }

    @SuppressWarnings("serial")
    private static class EventTSWAssigner extends AscendingTimestampExtractor<Event> {

        @Override
        public long extractAscendingTimestamp(Event element) {
            return element.getTimestamp();
        }
    }

    @SuppressWarnings("serial")
    private static class SortByTimestampFunction extends KeyedProcessFunction<String, Event, Event> {
        private ValueState<PriorityQueue<Event>> queueState = null;

        @Override
        public void open(Configuration config) {
            ValueStateDescriptor<PriorityQueue<Event>> descriptor = new ValueStateDescriptor<>(
                    // state name
                    "sorted-events",
                    // type information of state
                    TypeInformation.of(new TypeHint<PriorityQueue<Event>>() {
                    }));
            queueState = getRuntimeContext().getState(descriptor);
        }

        @Override
        public void processElement(Event event, Context context, Collector<Event> out) throws Exception {
            TimerService timerService = context.timerService();

            long currentWatermark = timerService.currentWatermark();
            System.out.format("processElement called with watermark %d\n", currentWatermark);
            if (context.timestamp() > currentWatermark) {
                PriorityQueue<Event> queue = queueState.value();
                if (queue == null) {
                    queue = new PriorityQueue<>(10);
                }

                queue.add(event);
                queueState.update(queue);
                timerService.registerEventTimeTimer(event.getTimestamp());
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext context, Collector<Event> out) throws Exception {
            PriorityQueue<Event> queue = queueState.value();
            long watermark = context.timerService().currentWatermark();
            System.out.format("onTimer called  with watermark %d\n", watermark);
            Event head = queue.peek();
            while (head != null && head.getTimestamp() <= watermark) {
                out.collect(head);
                queue.remove(head);
                head = queue.peek();
            }
        }
    }

    @SuppressWarnings("serial")
    private static class EventSource extends RichParallelSourceFunction<Event> {

        private String _prefix;

        private transient Random _rand;
        private transient boolean _running;
        private transient int _numEvents;

        public EventSource(String prefix) {
            _prefix = prefix;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);

            _rand = new Random(_prefix.hashCode() + getRuntimeContext().getIndexOfThisSubtask());
        }

        @Override
        public void cancel() {
            _running = false;
        }

        @Override
        public void run(SourceContext<Event> context) throws Exception {
            _running = true;
            _numEvents = 0;
            long timestamp = System.currentTimeMillis() + _rand.nextInt(10);

            while (_running && (_numEvents < 100)) {
                long deltaTime = timestamp - System.currentTimeMillis();
                if (deltaTime > 0) {
                    Thread.sleep(deltaTime);
                }

                context.collect(new Event(_prefix, timestamp));
                _numEvents++;

                // Generate a timestamp every 5...15 ms, average is 10.
                timestamp += (5 + _rand.nextInt(10));
            }
        }

    }
}