Hi Soheil, Hequn is right. This might be an issue with advancing event-time. You can monitor that by checking the watermarks in the web dashboard or print-debug it with a ProcessFunction which can lookup the current watermark.
Best, Fabian 2018-07-19 3:30 GMT+02:00 Hequn Cheng <chenghe...@gmail.com>: > Hi Soheil, > > > wait 8 milliseconds (according to my code) to see if any other data > with the same key received or not and after 8 millisecond it will be > triggered. > Yes, but the time is event time, so if there is no data from source the > time won't be advanced. > > There are some reasons why the event time has not been advanced: > 1. There are no data from the source > 2. One of the source parallelisms doesn't have data > 3. The time field, i.e, Long in Tuple3, should be millisecond instead of > second. > 4. Data should cover a longer time spam than the window size to advance > the event time. > > Best, Hequn > > On Wed, Jul 18, 2018 at 3:53 PM, Soheil Pourbafrani <soheil.i...@gmail.com > > wrote: > >> Hi, >> >> In a datastream processing problem, the source generated data every 8 >> millisecond and timestamp is a field of the data. In default Flink time >> behavior data enter the time window but when I set Flink time to EventTime >> it will output nothing! Here is the code: >> >> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); >> >> SingleOutputStreamOperator<Tuple3<String,Long, JSONObject>> res = >> aggregatedTuple >> .assignTimestampsAndWatermarks(new >> BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, Long, >> JSONObject>>(Time.milliseconds(8)) { >> >> @Override >> public long extractTimestamp(Tuple3<String, Long, JSONObject> >> element) { >> return element.f1 ; >> } >> }).keyBy(1).timeWindow(Time.milliseconds(8)) >> .allowedLateness(Time.milliseconds(3)) >> .sideOutputLateData(lateOutputTag) >> .reduce(processing...); >> DataStream<Tuple3<String, Long, JSONObject>> lateData = >> res.getSideOutput(lateOutputTag); >> res.print(); >> >> What is the problem with my code? >> According to the Flink documents, my understanding about EventTime is >> that for example in case of time window when a new data received it start a >> new (logical window) based on new data event timestamp and wait 8 >> milliseconds (according to my code) to see if any other data with the same >> key received or not and after 8 millisecond (from timestamp of the first >> element of the window) it will be triggered. Since data source generated >> data in a constant periodic interval, I set a watermarck of 8, too. Is my >> understanding about Flink window in EventTime correct? >> > >