Flink自定义触发器开发指南

触发器的作用

1. 监控进入窗口的数据元素。

2. 根据时间，计数或其他业务条件判断是否应触发窗口。

3. 返回以下操作之一：

   • CONTINUE：继续收集数据。

   • FIRE：触发窗口计算并保留数据；

   • FIRE_AND_PURGE：触发窗口计算并清除数据；

   • PURGE：仅清除数据而不触发计算。

内置触发器类型

Flink 提供了一些常用的内置触发器，适用于基本的时间或计数驱动场景：

虽然内置触发器能满足一些基本需求，但在实际应用中，我们常常需要：

• 组合多个触发条件 （如“5条记录或1分钟超时”）；

• 基于特定事件触发 （如用户登出、订单完成）；

• 控制窗口生命周期 （早发、延迟、多次触发等）；

• 避免默认行为带来的副作用 （如防止迟到数据再次触发窗口）；

这些复杂需求无法通过简单的内置触发器实现，因此引入自定义触发器成为必要选择。

场景示例：在一个滚动事件时间窗口（如每小时）中，若某用户活动频繁，则在其进入窗口的第5条记录时立即触发一次窗口计算，同时仍保证在窗口结束时再触发一次最终结果。

代码示例：数据达到五条需要进行一次窗口计算，且只额外触发一次。

public class CustomCountTrigger extends Trigger<UserEvent, TimeWindow> {

    // 用于记录每个 key 和 window 中已经接收的元素数量
    private final ValueStateDescriptor<Integer> countStateDesc =
            new ValueStateDescriptor<>("count", Integer.class);
    // 用于记录是否已经触发过计算
    private final ValueStateDescriptor<Boolean> flagStateDesc =
            new ValueStateDescriptor<>("flag", Boolean.class);

    // 当有一个新元素进入窗口时调用
    @Override
    public TriggerResult onElement(UserEvent event, long timestamp, TimeWindow window, Trigger.TriggerContext ctx) throws Exception {
        // 获取当前 key + window 对应的状态值
        ValueState<Integer> countState = ctx.getPartitionedState(countStateDesc);
        ValueState<Boolean> flagState = ctx.getPartitionedState(flagStateDesc);
        int count = countState.value() == null ? 0 : countState.value();
        boolean flag = flagState.value() == null ? false : flagState.value();

        // 每来一条数据，计数加一
        count += 1;
        countState.update(count); // 更新状态
        
        // 如果计数 = 5，立即触发窗口计算
        if (count >= 5 && !flag) {
            flagState.update(true); // 更新状态，保证只触发一次
            return TriggerResult.FIRE;
        } else {
            return TriggerResult.CONTINUE;
        }
    }

    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext triggerContext) throws Exception {
        // 处理时间定时器不处理 
        return TriggerResult.CONTINUE;
    }

    // 当注册的事件时间定时器被触发时调用（例如窗口结束时间）
    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.FIRE_AND_PURGE; // 触发并清除窗口
    }

    @Override
    public void clear(TimeWindow timeWindow, TriggerContext ctx) throws Exception {
        // 清理窗口状态
        ctx.getPartitionedState(countStateDesc).clear();
        ctx.getPartitionedState(flagStateDesc).clear();
    }
}

场景一

如果想多次告警，触发窗口内的计算和事件，应该如何处理？

1. 去除flagStateDesc标志位，即可在多次触发计算和事件。

2. 如果也可以添加计数标记来达到触发一定次数后再结束。（如告警事件）

   ### 去除标志位后，如有八条数据则会产生5条记录。  
   Key: 101, 访问次数为：5       // count >= 5,提前触发计算，不清空状态。
   Key: 101, 访问次数为：6       // 满足count >= 5,提前触发计算
   Key: 101, 访问次数为：7      // 满足count >= 5,提前触发计算
   Key: 101, 访问次数为：8      // 满足count >= 5,提前触发计算
   Key: 101, 访问次数为：8      //  窗口关闭，触发计算，清空状态。

3. 触发器只决定何时触发计算，获取真正的对象信息的内容在.process当中，可以通过数量和条件状态等，输出不同的collect。

场景二

窗口的关闭是需要水平线推动的，如果在很长一段时间内没有数据产生，如何保证能按时触发关窗计算呢？

• 方案一：改用 ProcessingTime 窗口

  如果不需要事件时间语义（比如不关心事件发生的时间），可以直接使用处理时间窗口：

  .keyBy(keySelector)
  .window(TumblingProcessingTimeWindows.of(Time.seconds(60))) // 使用处理时间
  .process(new MyProcessWindowFunction())

• 方案二：使用 withIdleness 方法

  Flink的WatermarkStrategy 提供了一个便捷方法 withIdleness，用于检测数据源是否处于空闲状态，并在空闲超过指定时间后将其标记为空闲，从而避免其阻塞整体 Watermark 的生成。

  // 空闲的数据源将不再参与 Watermark 的最小值计算，从而不会阻碍其他活跃数据源的 Watermark 推进。
  WatermarkStrategy
      .<Tuple2<Long, String>>forBoundedOutOfOrderness(Duration.ofSeconds(20))
      .withIdleness(Duration.ofMinutes(1));  //表示如果一个数据源或分区在 1 分钟内没有任何事件，则被标记为空闲

• 方案三：自定义 WatermarkGenerator

  如果坚持使用事件时间，就需要确保即使没有新数据到达，也能继续推进水位线 。可以实现一个自定义生成Watermark的机制。

  • 在onEvent()方法中记录最近一次接收到事件的时间。

  • 在onPeriodicEmit()方法中检查当前时间与最后一次接收到事件的时间之间的间隔。

  • 如果间隔超过了设定的阈值，则认为该数据源已空闲，跳过生成 Watermark 或直接生成特定的Watermark。

  public class IdleAwareWatermarkGenerator implements WatermarkGenerator<MyEvent> {
      private long lastEventTimestamp = Long.MIN_VALUE;
      private final long maxIdleTimeMs; // 最大空闲时间
  
      public IdleAwareWatermarkGenerator(long maxIdleTimeMs) {
          this.maxIdleTimeMs = maxIdleTimeMs;
      }
  
      @Override
      public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
          lastEventTimestamp = Math.max(lastEventTimestamp, eventTimestamp);
      }
  
      @Override
      public void onPeriodicEmit(WatermarkOutput output) {
          long currentTime = System.currentTimeMillis();
          if (lastEventTimestamp == Long.MIN_VALUE || currentTime - lastEventTimestamp > maxIdleTimeMs) {
              // 如果长时间没有事件，则不发出新的 Watermark
              return;
          }
          output.emitWatermark(new Watermark(lastEventTimestamp));
      }
  }

  WatermarkStrategy<MyEvent> strategy = WatermarkStrategy
      .forGenerator((ctx) -> new IdleAwareWatermarkGenerator(60_000)) // 设置最大空闲时间为 60 秒
      .withTimestampAssigner((event, timestamp) -> event.getEventTime());

• 方案三：强制注册定时器

  @Override
  public TriggerResult onElement(Event event, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
      ctx.registerEventTimeTimer(window.maxTimestamp()); // 基于事件时间
      ctx.registerProcessingTimeTimer(window.maxTimestamp() + 1000); // 容错：即使没事件也触发
      return TriggerResult.CONTINUE;
  }
  
  @Override
  public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
      return TriggerResult.FIRE_AND_PURGE; // 强制触发并清空
  }

• 方案四：外部维护心跳器

  可以通过Kafka或者任务编排，定时向数据下游发送心跳数据，触发关窗操作。

核心要点总结