Flink事件时间处理水印总是-9223372036854725808

Question

我试图使用process函数对一组事件进行处理。我正在使用事件时间和键控流。我面临的问题是，水印值始终为9223372036854725808。我已经对print语句进行了调试，它显示如下：

时间戳------1583128014000提取时间戳1583128014000当前水印------9223372036854775808

时间戳------1583128048000提取时间戳1583128048000当前水印------9223372036854775808

时间戳------158312808900提取的时间戳158312808900当前水印------9223372036854775808

所以时间戳和提取的时间戳会改变，但水印不会更新。所以没有记录排队，因为context.timestamp永远不会少于水印。

DataStream<GenericRecord> dataStream = env.addSource(searchConsumer).name("search_list_keyless");
        DataStream<GenericRecord> dataStreamWithWaterMark =  dataStream.assignTimestampsAndWatermarks(new SessionAssigner());

       try {
            dataStreamWithWaterMark.keyBy((KeySelector<GenericRecord, String>) record -> {
                StringBuilder builder = new StringBuilder();
                builder.append(record.get("session_id"));
                builder.append(record.get("user_id"));
                return builder.toString();
            }).process(new MatchFunction()).print();
        }
        catch (Exception e){
            e.printStackTrace();
        }
        env.execute("start session process");

    }

    public static class SessionAssigner implements AssignerWithPunctuatedWatermarks<GenericRecord>  {
        @Override
        public long extractTimestamp(GenericRecord record, long previousElementTimestamp) {
            long timestamp = (long) record.get("event_ts");
            System.out.println("timestamp------"+ timestamp);
            return timestamp;
        }

        @Override
        public Watermark checkAndGetNextWatermark(GenericRecord record, long extractedTimestamp) {
            // simply emit a watermark with every event
            System.out.println("extractedTimestamp "+extractedTimestamp);
            return new Watermark(extractedTimestamp - 30000);
        }
 }

这是processFunction的代码。。。。

public class MatchFunction extends KeyedProcessFunction<String, GenericRecord, Object> {

    private ValueState<Tuple2<Long, PriorityQueue<GenericRecord>>> queueState = null;

    @Override
    public void open(Configuration config) throws Exception {
        System.out.println("open");
        ValueStateDescriptor<Tuple2<Long, PriorityQueue<GenericRecord>>> descriptor = new ValueStateDescriptor<>(
                "sorted-events", TypeInformation.of(new TypeHint<Tuple2<Long, PriorityQueue<GenericRecord>>>() {
        })
        );
        queueState = getRuntimeContext().getState(descriptor);
    }
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Object> out) throws Exception {
        Tuple2<Long, PriorityQueue<GenericRecord>> tuple = queueState.value();

        PriorityQueue<GenericRecord> records = tuple.f1;

    }

    @Override
    public void processElement(GenericRecord record, Context context, Collector<Object> collector) throws Exception {

        TimerService timerService = context.timerService();
        System.out.println("currentwatermark----"+ timerService.currentWatermark());
        if (context.timestamp() > timerService.currentWatermark()) {

            Tuple2<Long, PriorityQueue<GenericRecord>> queueval = queueState.value();
            PriorityQueue<GenericRecord> queue = queueval.f1;
            long startTime = queueval.f0;
            System.out.println("starttime----"+ startTime);

            if (queue == null) {
                queue = new PriorityQueue<>(10, new TimeStampComprator());
                startTime = (long) record.get("event_ts");
            }
            queueState.update(new Tuple2<>(startTime, queue));
            timerService.registerEventTimeTimer(startTime + 5 * 60 * 1000);
        }
    }

}

Answer 1

以下是你分享的一个可能的解释：

时间戳和标点水印操作符在调用给定记录的checkAndGetNextWatermark之前调用提取时间戳。这意味着在每个任务（并行实例）中第一次调用匹配函数中的processElement时，当前水印将很长。最小值（即-9223372036854775808）。

如果你的平行度足够大，那就可以解释看到

currentwatermark-----9223372036854775808

好几次。