flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Biplob Biswas <revolutioni...@gmail.com>
Subject Re: Flink CEP not emitting timed out events properly
Date Tue, 20 Jun 2017 14:18:24 GMT
Hi Kostas,

Yes, I have a flag in my timestampextractor.

As you can see from the code below, I am checking whether 
currentTime - systemTimeSinceLastModification > 10 sec..... as new events
come then the watermark wouldn't be incremented. But as soon as I have a
difference of more than 10 seconds, I am incrementing the watermark by 1
sec, I feel this is very small and I would try incrementing the watermark
with a higher value but yeah this is what I am doing.


public class TimestampAndWatermarkGenerator implements
AssignerWithPeriodicWatermarks<BAMEvent>{

  private final long maxOutOfOrderness = 10000; // 10 seconds

  private long currentMaxTimestamp;
  private long systemTimeSinceLastModification;
  private boolean firstEventFlag = false;
  private Logger log =
LoggerFactory.getLogger(TimestampAndWatermarkGenerator.class);

  @Nullable
  @Override
  public Watermark getCurrentWatermark() {
    long currentTime = System.currentTimeMillis();
    if(firstEventFlag && (currentTime - systemTimeSinceLastModification >
10000)){
      systemTimeSinceLastModification = currentTime;
      currentMaxTimestamp = currentMaxTimestamp + 1000;
      //log.info("Current Max Time - {}, Last Modification Time - {}",
currentMaxTimestamp, systemTimeSinceLastModification );
    }
    return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
  }

  @Override
  public long extractTimestamp(BAMEvent bamEvent, long l) {
    long timestamp = bamEvent.getTimestamp();
    currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
    systemTimeSinceLastModification = System.currentTimeMillis();
    firstEventFlag = true;
    //log.info("Current Max Time - {}, Current Event Time - {}",
currentMaxTimestamp, systemTimeSinceLastModification);

    return timestamp;
  }
}



--
View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-CEP-not-emitting-timed-out-events-properly-tp13794p13860.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.

Mime
View raw message