CODEWITHSUNDEEP
javaapache-flink
Cihan Kara
Cihan Kara

Reputation: 47

Event Time Based Window Doesn't Fire

I am working on Flink event time based window. But when i send kafka message program does not do windowing operation. I did everything what docs say but couldnt solve the problem any help will be appriciated, thanks in advance

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.getConfig();
        environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        environment.setParallelism(1);
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id","event-group");

        FlinkKafkaConsumer<EventSalesQuantity> consumer = new FlinkKafkaConsumer<EventSalesQuantity>("EventTopic",new EventSerializationSchema(),props);
        DataStream<EventSalesQuantity> eventDataStream = environment.addSource(consumer);

        KeyedStream<EventSalesQuantity, String> keyedEventStream = eventDataStream.assignTimestampsAndWatermarks( new AssignerWithPeriodicWatermarksImpl()).
           keyBy(new KeySelector<EventSalesQuantity, String>() {
               @Override
               public String getKey(EventSalesQuantity eventSalesQuantity) throws Exception {
                   return  eventSalesQuantity.getDealer();
               }
           });

        DataStream<Tuple2<EventSalesQuantity,Integer>> eventSinkStream = keyedEventStream.timeWindow(Time.seconds(5)).aggregate(new AggregateImpl());
        eventSinkStream.addSink(new FlinkKafkaProducer<Tuple2<EventSalesQuantity, Integer>>("localhost:9092","SinkEventTopic",new EventSinkSerializationSchema()));
        eventSinkStream.print();
        environment.execute();
    }
}




public class AssignerWithPeriodicWatermarksImpl implements AssignerWithPeriodicWatermarks<EventSalesQuantity> {
    private final long maxOutOfOrderness = 3500; 
    private long currentMaxTimestamp;

    @Override
    public long extractTimestamp(EventSalesQuantity element, long previousElementTimestamp) {
        long timestamp = DateUtils.getDateFromString(element.getTransactionDate()).getTime();
        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
        return timestamp;
    }

    @Override
    public Watermark getCurrentWatermark() {
        // return the watermark as current highest timestamp minus the out-of-orderness bound
        return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
    }

"2019-06-21T09:43:01" "2019-06-21T09:43:03"

I send 2 messages with these time stamps but i got no output.

Upvotes: 1

Views: 146

Answers (1)

David Anderson
David Anderson

Reputation: 43499

  1. Your event time windows are 5 seconds long. The window containing those events won't be triggered until it sees a watermark with a timestamp of at least 2019-06-21T09:43:05.
  2. With the maxOutOfOrderness set to 3500 msec, your watermark generator won't generate a watermark large enough to trigger the window until it has seen an event with a timestamp of at least 2019-06-21T09:43:08.500.

Upvotes: 4

Related Questions