Flink 自定义Trigger实战代码，结合State来使用

浅浅的花香味﹌ 2023-05-29 13:05 45阅读 0赞

简单说明： 比如window大小为20秒，在进入window的时候 有一个valueState在统计数据数量，数量大于等于2的时候直接触发这个窗口，如果在20秒事件窗口内只有一条数据进入 那么正常等window事件到了 就会触发这一条数据了 。

>     package appOperator;
>     
>     import com.alibaba.fastjson.JSONObject;
>     import org.apache.flink.api.common.state.ValueState;
>     import org.apache.flink.api.common.state.ValueStateDescriptor;
>     import org.apache.flink.streaming.api.windowing.triggers.Trigger;
>     import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
>     import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
>     
>     public class MyTrigger extends Trigger<JSONObject, TimeWindow> {
>         private static final long serialVersionUID = 1L;
>         ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("total", Integer.class);
>         private static int flag = 0;
>         
>         private TriggerResult fireAndPurge(TimeWindow window, TriggerContext ctx) throws Exception {
>             clear(window, ctx);
>             return TriggerResult.FIRE_AND_PURGE;
>         }
>     
>     
>         @Override
>         public TriggerResult onElement(JSONObject element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
>     
>             //todo 注册时间
>             ctx.registerProcessingTimeTimer(window.maxTimestamp());
>             ValueState<Integer> sumState = ctx.getPartitionedState(stateDescriptor);
>     
>     
>            /* if(flag >= 2){
>                 flag = 0;
>                 return TriggerResult.FIRE;
>             }else{
>                 flag++;
>             }
>     */
>             if (null == sumState.value()) {
>                 sumState.update(0);
>             }
>             int newValue = sumState.value()+1 ;
>             sumState.update(newValue);
>             System.out.println("newValue = "+newValue+",sumState.value() = "+sumState.value());
>             if (sumState.value() >= 2) {
>                 System.out.println("进入。。。。。。。。。。。。");
>                 return fireAndPurge(window, ctx);
>             }
>     
>             return TriggerResult.CONTINUE ;
>     
>         }
>     
>         @Override
>         public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
>     
>             return TriggerResult.FIRE;
>     
>         }
>     
>         @Override
>         public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
>     
>             return TriggerResult.CONTINUE;
>     
>      
>         }
>     
>         @Override
>         public void onMerge(TimeWindow window,
>                             OnMergeContext ctx) {
>     
>             long windowMaxTimestamp = window.maxTimestamp();
>             if (windowMaxTimestamp > ctx.getCurrentProcessingTime()) {
>                 ctx.registerProcessingTimeTimer(windowMaxTimestamp);
>             }
>         }
>     
>         @Override
>         public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
>             ValueState<Integer> partitionedState = ctx.getPartitionedState(stateDescriptor);
>             partitionedState.clear();
>             ctx.deleteProcessingTimeTimer(window.maxTimestamp());
>         }
>     }

调用的代码：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191106114353535.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz,size_16,color_FFFFFF,t_70