Hi Hequn, To my understand, a processing time window is fired at the last millisecond of the window(maxTimestamp). Then what will happen if more elements arrive at the last millisecond, but AFTER the window is fired?
Thanks, Youjun 发件人: Hequn Cheng <chenghe...@gmail.com> 发送时间: Friday, July 13, 2018 9:44 PM 收件人: Yuan,Youjun <yuanyou...@baidu.com> 抄送: Timo Walther <twal...@apache.org>; user@flink.apache.org 主题: Re: 答复: 答复: TumblingProcessingTimeWindow emits extra results for a same window Hi Youjun, The rowtime value in udf:EXTRACT(EPOCH FROM rowtime) is different from the rowtime value of window. Sql will be parsed and translated into some nodes, Source -> Calc -> Window -> Sink. The Calc is the input node of Window and the udf is part of Calc instead of Window. So the max_ts and min_ts is actually the time before entering the window, i.e, not the time in window. However, I still can't find anything valuable to solve the problem. It seems the window has been triggered many times for the same key. I will think more about it. Best, Hequn. On Fri, Jul 13, 2018 at 11:53 AM, Yuan,Youjun <yuanyou...@baidu.com<mailto:yuanyou...@baidu.com>> wrote: Hi Hequn, I am using Flink 1.4. The job was running with 1 parallelism. I don’t think the extra records are caused by different keys, because: 1. I ran 2 jobs consuming the same source, jobA with 2-minute window, and job with 4-minute window. If there are wired keys, then jobA will get no more records than jobB, for the same period. But that not true, jobA got 17 records while jobB got 11. Relevant results could be found below. 2. For each window, I output the min and max timestamp, and found that those extra records always start at the last few milliseconds of the 2 or 4-minte windows, just before window got closed. I also noticed the windows did not have a clear cut between minutes, as we can see in jobA’s output, ts 1531448399978 appears in 18 result records, either as start, or end, or both. jobA(2-minute window) output {"timestamp":1531448040000,"cnt":1668052,"userId":"user01","min_ts":1531448040003,"max_ts":1531448159985} {"timestamp":1531448160000,"cnt":1613188,"userId":"user01","min_ts":1531448159985,"max_ts":1531448279979} {"timestamp":1531448280000,"cnt":1664652,"userId":"user01","min_ts":1531448280004,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":3,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":2,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":4,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":3,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":2,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":2,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":2,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":3,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":2,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":2,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":3,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":3,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":3,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":2,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":3,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448280000,"cnt":3,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448400000,"cnt":1593435,"userId":"user01","min_ts":1531448399978,"max_ts":1531448519978} jobB(4-minute window) output {"timestamp":1531447920000,"cnt":3306838,"userId":"user01","min_ts":1531447919981,"max_ts":1531448159975} {"timestamp":1531448160000,"cnt":3278178,"userId":"user01","min_ts":1531448159098,"max_ts":1531448399977} {"timestamp":1531448160000,"cnt":4,"userId":"user01","min_ts":1531448399977,"max_ts":1531448399977} {"timestamp":1531448160000,"cnt":5,"userId":"user01","min_ts":1531448399977,"max_ts":1531448399977} {"timestamp":1531448160000,"cnt":8,"userId":"user01","min_ts":1531448399977,"max_ts":1531448399978} {"timestamp":1531448160000,"cnt":7,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448160000,"cnt":2,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448160000,"cnt":3,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448160000,"cnt":3,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448160000,"cnt":3,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448160000,"cnt":3,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448160000,"cnt":3,"userId":"user01","min_ts":1531448399978,"max_ts":1531448399978} {"timestamp":1531448400000,"cnt":3226735,"userId":"user01","min_ts":1531448399978,"max_ts":1531448639916} Thanks Youjun 发件人: Hequn Cheng <chenghe...@gmail.com<mailto:chenghe...@gmail.com>> 发送时间: Thursday, July 12, 2018 11:31 PM 收件人: Yuan,Youjun <yuanyou...@baidu.com<mailto:yuanyou...@baidu.com>> 抄送: Timo Walther <twal...@apache.org<mailto:twal...@apache.org>>; user@flink.apache.org<mailto:user@flink.apache.org> 主题: Re: 答复: TumblingProcessingTimeWindow emits extra results for a same window Hi Yuan, Haven't heard about this before. Which flink version do you use? The cause may be: 1. userId not 100% identical, for example contains invisible characters. 2. The machine clock vibrated. Otherwise, there are some bugs we don't know. Best, Hequn On Thu, Jul 12, 2018 at 8:00 PM, Yuan,Youjun <yuanyou...@baidu.com<mailto:yuanyou...@baidu.com>> wrote: Hi Timo, This problem happens 4-5 times a day on our online server, with ~15k events per second load, and it is using PROCESSING time. So I don’t think I can stably reproduce the issue on my local machine. The user ids are actually the same, I have doubled checked that. Now, I am wondering could it possible that, after a window fires, some last events came but that still fall to the time range of the just fired window, hence new windows are assigned, and fired later. This can explain why the extra records always contain only a few events (cnt is small). To verify that, I just modified the SQL to also output the MIN timestamp of each windows, and I found the MIN timestamp of the extra records are always point to the LAST second of the window. Here is the output I just got, note 1531395119 is the last second of a 2-minute window start from 1531395000. -------------------------------------------------------------------------------------------------------------------------------- {"timestamp":1531394760000,"cnt":1536013,"userId":"user01","min_sec":1531394760} {"timestamp":1531394880000,"cnt":1459623,"userId":"user01","min_sec":1531394879} {"timestamp":1531395000000,"cnt":1446010,"userId":"user01","min_sec":1531395000} {"timestamp":1531395000000,"cnt":7,"userId":"user01","min_sec":1531395119} {"timestamp":1531395000000,"cnt":3,"userId":"user01","min_sec":1531395119} {"timestamp":1531395000000,"cnt":3,"userId":"user01","min_sec":1531395119} {"timestamp":1531395000000,"cnt":6,"userId":"user01","min_sec":1531395119} {"timestamp":1531395000000,"cnt":3,"userId":"user01","min_sec":1531395119} {"timestamp":1531395000000,"cnt":2,"userId":"user01","min_sec":1531395119} {"timestamp":1531395000000,"cnt":2,"userId":"user01","min_sec":1531395119} {"timestamp":1531395000000,"cnt":2,"userId":"user01","min_sec":1531395119} The modified SQL: INSERT INTO sink SELECT TUMBLE_START(rowtime, INTERVAL '2' MINUTE) AS `timestamp`, count(vehicleId) AS cnt, userId, MIN(EXTRACT(EPOCH FROM rowtime)) AS min_sec FROM source GROUP BY TUMBLE(rowtime, INTERVAL '2' MINUTE), userId thanks Youjun 发件人: Timo Walther <twal...@apache.org<mailto:twal...@apache.org>> 发送时间: Thursday, July 12, 2018 5:02 PM 收件人: user@flink.apache.org<mailto:user@flink.apache.org> 主题: Re: TumblingProcessingTimeWindow emits extra results for a same window Hi Yuan, this sounds indeed weird. The SQL API uses regular DataStream API windows underneath so this problem should have come up earlier if this is problem in the implementation. Does this behavior reproducible on your local machine? One thing that comes to my mind is that the "userId"s might not be 100% identical (same hashCode/equals method) because otherwise they would be properly grouped. Regards, Timo Am 12.07.18 um 09:35 schrieb Yuan,Youjun: Hi community, I have a job which counts event number every 2 minutes, with TumblingWindow in ProcessingTime. However, it occasionally produces extra DUPLICATED records. For instance, for timestamp 1531368480000 below, it emits a normal result (cnt=1641161), and then followed by a few more records with very small result (2, 3, etc). Can anyone shed some light on the possible reason, or how to fix it? Below are the sample output. ----------------------------------------------------------- {"timestamp":1531368240000,"cnt":1537821,"userId":"user01"} {"timestamp":1531368360000,"cnt":1521464,"userId":"user01"} {"timestamp":1531368480000,"cnt":1641161,"userId":"user01"} {"timestamp":1531368480000,"cnt":2,"userId":"user01"} {"timestamp":1531368480000,"cnt":3,"userId":"user01"} {"timestamp":1531368480000,"cnt":3,"userId":"user01"} And here is the job SQL: ----------------------------------------------------------- INSERT INTO sink SELECT TUMBLE_START(rowtime, INTERVAL '2' MINUTE) AS `timestamp`, count(vehicleId) AS cnt, userId FROM source GROUP BY TUMBLE(rowtime, INTERVAL '2' MINUTE), userId Thanks, Youjun Yuan
