---------- Forwarded message ---------
From: 远远 <[email protected]>
Date: 2018年9月16日周日 下午4:08
Subject: Re: why same Sliding ProcessTime TimeWindow triggered twice
To: <[email protected]>
hi, the flink version that i test is 1.4.2, and i just run test code with
local env in IDEA, and all the setting in the test code.
my os is deepin(linux based debian) 15.7...
and i try again, the print as flow:
now ===> 2018-09-16 16:06:09
start ===> 2018-09-16 16:05:10
end ===> 2018-09-16 16:06:10
max ===> 2018-09-16 16:06:09
TimeWindow{start=1537085110000, end=1537085170000}
aggreation
now ===> 2018-09-16 16:06:09
start ===> 2018-09-16 16:05:10
end ===> 2018-09-16 16:06:10
max ===> 2018-09-16 16:06:09
TimeWindow{start=1537085110000, end=1537085170000}
aggreation
now ===> 2018-09-16 16:06:16
start ===> 2018-09-16 16:05:15
end ===> 2018-09-16 16:06:15
max ===> 2018-09-16 16:06:14
TimeWindow{start=1537085115000, end=1537085175000}
aggreation
now ===> 2018-09-16 16:06:19
start ===> 2018-09-16 16:05:20
end ===> 2018-09-16 16:06:20
max ===> 2018-09-16 16:06:19
TimeWindow{start=1537085120000, end=1537085180000}
aggreation
now ===> 2018-09-16 16:06:20
start ===> 2018-09-16 16:05:20
end ===> 2018-09-16 16:06:20
max ===> 2018-09-16 16:06:19
TimeWindow{start=1537085120000, end=1537085180000}
aggreation
now ===> 2018-09-16 16:06:24
start ===> 2018-09-16 16:05:25
end ===> 2018-09-16 16:06:25
max ===> 2018-09-16 16:06:24
TimeWindow{start=1537085125000, end=1537085185000}
aggreation
now ===> 2018-09-16 16:06:24
start ===> 2018-09-16 16:05:25
end ===> 2018-09-16 16:06:25
max ===> 2018-09-16 16:06:24
TimeWindow{start=1537085125000, end=1537085185000}
aggreation
now ===> 2018-09-16 16:06:25
start ===> 2018-09-16 16:05:25
end ===> 2018-09-16 16:06:25
max ===> 2018-09-16 16:06:24
TimeWindow{start=1537085125000, end=1537085185000}
aggreation
now ===> 2018-09-16 16:06:29
start ===> 2018-09-16 16:05:30
end ===> 2018-09-16 16:06:30
max ===> 2018-09-16 16:06:29
TimeWindow{start=1537085130000, end=1537085190000}
aggreation
now ===> 2018-09-16 16:06:29
start ===> 2018-09-16 16:05:30
end ===> 2018-09-16 16:06:30
max ===> 2018-09-16 16:06:29
TimeWindow{start=1537085130000, end=1537085190000}
aggreation
now ===> 2018-09-16 16:06:30
start ===> 2018-09-16 16:05:30
end ===> 2018-09-16 16:06:30
max ===> 2018-09-16 16:06:29
TimeWindow{start=1537085130000, end=1537085190000}
aggreation
now ===> 2018-09-16 16:06:36
start ===> 2018-09-16 16:05:35
end ===> 2018-09-16 16:06:35
max ===> 2018-09-16 16:06:34
TimeWindow{start=1537085135000, end=1537085195000}
Xingcan Cui <[email protected]> 于2018年9月16日周日 下午3:55写道：
> Hi,
>
> I’ve tested your code in my local environment and everything worked fine.
> It’s a little weird to see your output like that. I wonder if you could
> give more information about your environment, e.g., your flink version and
> execution settings.
>
> Thanks,
> Xingcan
>
> On Sep 16, 2018, at 3:19 PM, 远远 <[email protected]> wrote:
>
> hi，everyone:
> today, i test Sliding ProcessTime TimeWindow with print some merties. i
> find a same sliding window be printed twice, as fllower:
>
> now ===> 2018-09-16 15:11:44
> start ===> 2018-09-16 15:10:45
> end ===> 2018-09-16 15:11:45
> max ===> 2018-09-16 15:11:44
> TimeWindow{start=1537081845000, end=1537081905000}
> aggreation
> now ===> 2018-09-16 15:11:45
> start ===> 2018-09-16 15:10:45
> end ===> 2018-09-16 15:11:45
> max ===> 2018-09-16 15:11:44
> TimeWindow{start=1537081845000, end=1537081905000}
> aggreation
>
> but when i do some sum operator，it will not, i want to know why?
> thanks.
>
> my test code is:
>
> object SlidingProcessTimeWindowTest {
>
> def main(args: Array[String]): Unit = {
> val env = StreamExecutionEnvironment.getExecutionEnvironment
> env.addSource((context: SourceContext[String]) => {while(true)
> context.collect(new Random().nextInt(100) + ":FRI")})
> .keyBy(s => s.endsWith("FRI"))
> .timeWindow(Time.minutes(1), Time.seconds(5))
> .apply((e, w, iter, coll: Collector[String]) => {
> println("now ===> " + convert(DateTime.now().getMillis))
> println("start ===> " + convert(w.getStart))
> println("end ===> " + convert(w.getEnd))
> println("max ===> " + convert(w.maxTimestamp()))
> println(w)
> // var reduce: Long = 0
> // for(e <- iter){
> // reduce += e.substring(0, e.length - 4).toInt
> // }
> // println("reduce ==> " + reduce)
> coll.collect("aggreation")
> }).setParallelism(1).print().setParallelism(1)
>
> env.execute()
> }
>
> def convert(time: Long): String = {
> new DateTime(time).toString("yyyy-MM-dd HH:mm:ss")
> }
> }
>
>
>
