I am trying to use process function to some processing on a set of
events. I am using event time and keystream. The issue I am facing is The watermark value is always coming as 9223372036854725808. I have put
print statement to debug and it shows like this:
timestamp------1583128014000
extractedTimestamp 1583128014000
currentwatermark-----9223372036854775808
timestamp------1583128048000
extractedTimestamp 1583128048000
currentwatermark-----9223372036854775808
timestamp------1583128089000
extractedTimestamp 1583128089000
currentwatermark-----9223372036854775808
timestamp and extracted timestamp changing but watermark not
getting updated. So no record is getting in the queue as context.timestamp is
never less than the 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);
}
}
@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);
}
}
}
Please help me to underand what i am doing wrong.
--
Thanks & Regards,
Anuj Jain
Locked
signature.asc (849 bytes)