Flink EventTime 处理水印总是出现 -9223372036854725808 [英] Flink EventTime Processing Watermark is always coming as -9223372036854725808
问题描述
我正在尝试使用 process 函数对一组事件进行一些处理.我正在使用事件时间和键控流.我面临的问题是水印值总是以 9223372036854725808 的形式出现.我已将打印语句进行调试,它显示如下:
I am trying to use process function to some processing on a set of events. I am using event time and keyed stream. The issue i am facing is Watermark value is always coming as 9223372036854725808. i have put print statement to debug and it shows like this:
时间戳------1583128014000提取时间戳 1583128014000当前水印-----9223372036854775808
timestamp------1583128014000 extractedTimestamp 1583128014000 currentwatermark-----9223372036854775808
时间戳------1583128048000提取时间戳 1583128048000当前水印-----9223372036854775808
timestamp------1583128048000 extractedTimestamp 1583128048000 currentwatermark-----9223372036854775808
时间戳------1583128089000提取时间戳 1583128089000当前水印-----9223372036854775808
timestamp------1583128089000 extractedTimestamp 1583128089000 currentwatermark-----9223372036854775808
所以时间戳和提取的时间戳改变了,但水印没有更新.所以没有记录进入队列,因为上下文.时间戳永远不会小于水印.
So timestamp and extractedTimestamp changing but watermark not getting updated.So no record is getting in queue as context.timestamp is never less than watermark.
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的代码....
This is the code for 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);
}
}
}
推荐答案
以下是您分享的内容的可能解释:
Here's a possible explanation for what you've shared:
TimestampsAndPunctuatedWatermarksOperator
在为给定记录调用 checkAndGetNextWatermark
之前先调用 extractTimestamp
.这意味着第一次在每个任务(并行实例)中调用 MatchFunction
中的 processElement
时,当前水印将为 Long.MIN_VALUE(即 -9223372036854775808).
The TimestampsAndPunctuatedWatermarksOperator
calls extractTimestamp
before it calls checkAndGetNextWatermark
for a given record. This means that the first time the processElement
in your MatchFunction
is called in each task (parallel instance), the current watermark will be Long.MIN_VALUE (which is -9223372036854775808).
如果你的并行度足够大,那可以解释看到
If your parallelism is large enough, that could explain seeing
currentwatermark-----9223372036854775808
几次.
这篇关于Flink EventTime 处理水印总是出现 -9223372036854725808的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!