?????? ProcessWindowFunction??????clear??????????????????-v1.10.1

Wed, 26 Aug 2020 23:03:13 -0700 

????????????????????????10??????????????1??????.window(TumblingEventTimeWindows.of(Time.minutes(10))).trigger(ContinuousEventTimeTrigger.of(Time.minutes(1)))????????????????10????????????????????????????????????????????????????????????????????????????


------------------ ???????? ------------------
??????:                                                                         
                                               "user-zh"                        
                                                            
<[email protected]&gt;;
????????:&nbsp;2020??8??26??(??????) ????1:31
??????:&nbsp;"[email protected]"<[email protected]&gt;;

????:&nbsp;Re: ProcessWindowFunction??????clear??????????????????-v1.10.1



????????????????aggregate(AggregateFunction<T, ACC, V&gt; aggFunction,
ProcessWindowFunction<V, R, K, W&gt; windowFunction)??????
????????????????????????????????????????processWindowFunction????????????????????????????context??????????????state??????????

x <[email protected]&gt; ??2020??8??25?????? ????6:25??????

&gt;
&gt; 
????????????????????????????ProcessWindowFunction??clear????????????????????????????????????????????????????????????????????????????????env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)??????????
&gt; .window(TumblingEventTimeWindows.of(Time.days(1)))
&gt; .trigger(ContinuousEventTimeTrigger.of(Time.minutes(1)))
&gt; .evictor(TimeEvictor.of(Time.seconds(0), true))
&gt; .process(new ProcessWindowFunction[IN,OUT,KEY,TimeWindow]{
&gt; private var state: MapState[String,Boolean] = _
&gt; override def open
&gt; override def process
&gt; override def clear(ctx: Context): Unit = {
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
state.clear()
&gt; }
&gt; }

回复