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>
发送时间: Thursday, July 12, 2018 11:31 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 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
