flink自定义trigger详解

川长思鸟来 2022-11-12 02:31 312阅读 0赞

### 适用的场景解释: ###

\[1\]中有句话是这样的:

"其实，我们要实现基于事件时间的窗口随意输出，比如1000个元素触发一次输出，那么我们就可以通过修改这个触发器来实现。"

这句话的意思是，默认的自带的trigger一般是基于EventTime的。

那么这1000 个元素可能跨度是一小时，也可能跨度是两小时，对吧

但是显然默认的Trigger只能是盯着EventTime(时间戳)来决定是否触发计算，并不能根据元素个数进行触发。

**也就是说，默认的Trigger盯着的跨度是"时间差"。而不是"个数差"**

**讲人话就是:**

**①例如Flink的Trigger默认每隔一天输出统计数据，**

**②但是不支持默认每隔一千个订单输出统计数据。**

**但是注意这里的一千个统计数据可能超过一天，甚至超过一周，耗时可能不固定。**

**因为你想啊，代码都是要把逻辑写死的对吧？**

**一千个订单可能一开始耗时一周，后来耗时一个月。那程序要怎么根据变化的时间来锁定一千个订单触发一次？**

**显然做不到，这个时候我们就希望锁定"个数间隔"、“个数差”，这个时候就需要自定义Trigger**

### 官方文档说明: ###

下面是官方文档\[4\]中Triggers这一节的内容概括

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td style="width:200px;">需要override的函数</td> 
   <td style="width:298px;">函数作用</td> 
  </tr> 
  <tr> 
   <td style="width:200px;"><code>onElement()</code></td> 
   <td style="width:298px;">数据（element）被加入window的时候会调用该函数</td> 
  </tr> 
  <tr> 
   <td style="width:200px;"><code>onEventTime()</code>&nbsp;</td> 
   <td style="width:298px;"> <p>当一个注册的Event-Time定时器触发</p> </td> 
  </tr> 
  <tr> 
   <td style="width:200px;"><code>onProcessingTime()</code>&nbsp;</td> 
   <td style="width:298px;">当一个注册的Processing-Time定时器触发</td> 
  </tr> 
  <tr> 
   <td style="width:200px;"><code>onMerge()</code></td> 
   <td style="width:298px;"> <p>与有状态触发器(stateful triggers)和当两个窗口整合的时候整合(merge)状态相关。</p> <p>例如使用session windows</p> </td> 
  </tr> 
  <tr> 
   <td style="width:200px;"><code>clear()</code></td> 
   <td style="width:298px;">window清理数据需要</td> 
  </tr> 
 </tbody> 
</table>

前面三个用来设定调用事件(invocation event)以后如何操作，

所以这些"操作"必须是一个TriggerResult

也就是说，前三个函数返回的TriggerResult可以是下面几种选择:

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td><strong>返回的TriggerResult</strong></td> 
   <td><strong>作用</strong></td> 
  </tr> 
  <tr> 
   <td><code>CONTINUE</code></td> 
   <td>什么都不做</td> 
  </tr> 
  <tr> 
   <td><code>FIRE</code></td> 
   <td>触发计算</td> 
  </tr> 
  <tr> 
   <td><code>PURGE</code></td> 
   <td>删除窗口中的所有数据</td> 
  </tr> 
  <tr> 
   <td><code>FIRE_AND_PURG</code></td> 
   <td>触发计算后删除窗口中所有数据</td> 
  </tr> 
 </tbody> 
</table>

然后是Fire and Purge这一节的内容:

触发计算时，返回的一定是FIRE或者FIRE\_AND\_PURG(这个话仅仅是来自官方文档的翻译，其实Intellij提示的选项并不仅仅是上面几个)

**具体示范代码参考\[5\]即可**

private static Logger LOG = LoggerFactory.getLogger(CountTriggerWithTimeout.class);
     
        /**
         * 窗口最大数据量
         */
        private int maxCount;
        /**
         * event time / process time
         */
        private TimeCharacteristic timeType;
        /**
         * 用于储存窗口当前数据量的状态对象
         */
        private ReducingStateDescriptor<Long> countStateDescriptor =
                new ReducingStateDescriptor("counter", new Sum(), LongSerializer.INSTANCE);
     
     
        public CountTriggerWithTimeout(int maxCount, TimeCharacteristic timeType) {
     
            this.maxCount = maxCount;
            this.timeType = timeType;
        }
     
     
        private TriggerResult fireAndPurge(TimeWindow window, TriggerContext ctx) throws Exception {
            clear(window, ctx);
            return TriggerResult.FIRE_AND_PURGE;
        }
     
     
        @Override
        public TriggerResult onElement(T element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
            ReducingState<Long> countState = ctx.getPartitionedState(countStateDescriptor);
            countState.add(1L);
     
            if (countState.get() >= maxCount) {
                LOG.info("fire with count: " + countState.get());
                return fireAndPurge(window, ctx);
            }
            if (timestamp >= window.getEnd()) {
                LOG.info("fire with tiem: " + timestamp);
                return fireAndPurge(window, ctx);
            } else {
                return TriggerResult.CONTINUE;
            }
        }
     
        @Override
        public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            if (timeType != TimeCharacteristic.ProcessingTime) {
                return TriggerResult.CONTINUE;
            }
     
            if (time >= window.getEnd()) {
                return TriggerResult.CONTINUE;
            } else {
                LOG.info("fire with process tiem: " + time);
                return fireAndPurge(window, ctx);
            }
        }
     
        @Override
        public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            if (timeType != TimeCharacteristic.EventTime) {
                return TriggerResult.CONTINUE;
            }
     
            if (time >= window.getEnd()) {
                return TriggerResult.CONTINUE;
            } else {
                LOG.info("fire with event tiem: " + time);
                return fireAndPurge(window, ctx);
            }
        }
     
        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
            ReducingState<Long> countState = ctx.getPartitionedState(countStateDescriptor);
            countState.clear();
        }
     
        /**
         * 计数方法
         */
        class Sum implements ReduceFunction<Long> {
     
            @Override
            public Long reduce(Long value1, Long value2) throws Exception {
                return value1 + value2;
            }
        }
    }

**Reference:**

**\[1\][flink自定义trigger-实现窗口随意输出][flink_trigger-]**

**\[2\][Flink 自定义Trigger][Flink _Trigger]**

**\[3\][Flink 自定义trigger][Flink _trigger]**

**\[4\][flink官方文档-窗口][flink_-]**

**\[5\][Flink 自定义触发器][Flink]**

[flink_trigger-]: https://zhuanlan.zhihu.com/p/57939498
[Flink _Trigger]: https://blog.csdn.net/baifanwudi/article/details/102937381
[Flink _trigger]: https://blog.csdn.net/yidan7063/article/details/84837245
[flink_-]: https://ci.apache.org/projects/flink/flink-docs-stable/zh/dev/stream/operators/windows.html
[Flink]: https://blog.csdn.net/qq_31866793/article/details/97917175