(DEPRECATED) Apache Flink User Mailing List archive.

TumblingProcessingTimeWindow emits extra results for a same window

Classic

List

Threaded

10 messages Options

Yuan,Youjun

TumblingProcessingTimeWindow emits extra results for a same window

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"}

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

Timo Walther

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

Yuan,Youjun

答复: TumblingProcessingTimeWindow emits extra results for a same window

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":6,"userId":"user01","min_sec":1531395119}

{"timestamp":1531395000000,"cnt":3,"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 <[hidden email]>
发送时间: Thursday, July 12, 2018 5:02 PM
收件人: [hidden email]
主题: Re: TumblingProcessingTimeWindow emits extra results for a same window

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

Hequn Cheng

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 <[hidden email]> 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 <[hidden email]>
发送时间: Thursday, July 12, 2018 5:02 PM
收件人: [hidden email]
主题: 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

Yuan,Youjun

答复: 答复: TumblingProcessingTimeWindow emits extra results for a same window

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:

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.
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}