全局窗口自定义触发器上的 allowedLateness答案

【问题标题】：allowedLateness on Global Window custom trigger全局窗口自定义触发器上的 allowedLateness
【发布时间】：2020-02-19 16:07:13
【问题描述】：

我已经为我的事件流创建了一个自定义触发器和处理函数。

DataStream<DynamoDBRow> dynamoDBRows =
    sensorEvents
        .keyBy("id")
        .window(GlobalWindows.create())
        .trigger(new MyCustomTrigger())
        .allowedLateness(Time.minutes(1)) # Note
        .process(new MyCustomWindowProcessFunction());

我的触发器基于事件参数。一旦接收到事件结束信号，MyCustomWindowProcessFunction() 将应用于窗口元素。

@Slf4j
public class MyCustomTrigger extends Trigger<SensorEvent, GlobalWindow> {

  @Override
  public TriggerResult onElement(SensorEvent element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {

    if (element.isEventEnd() == true) {
      return TriggerResult.FIRE_AND_PURGE;
    }

    return TriggerResult.CONTINUE;
  }

  @Override
  public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
    return TriggerResult.CONTINUE;
  }

  @Override
  public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
    return TriggerResult.CONTINUE;
  }

  @Override
  public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {}
}

传感器数据可能很少，即使在触发器之后也可能出现。所以我添加了.allowedLateness(Time.minutes(1))，以确保在处理时不会错过这些事件。

就我而言，allowedLateness 不起作用。

翻阅文档后发现了这个

如何在 GlobalWindow 中包含 allowedLateness？

注意：我也试过设置环境时间特性

env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);

更新：20-02-2020

目前正在考虑以下方法。（目前还没有工作）

@Slf4j
public class JourneyTrigger extends Trigger<SensorEvent, GlobalWindow> {

  private final long allowedLatenessMillis;

  public JourneyTrigger(Time allowedLateness) {
    this.allowedLatenessMillis = allowedLateness.toMilliseconds();
  }

  @Override
  public TriggerResult onElement(SensorEvent element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {

    if (element.isEventEnd() == true) {
      log.info("Timer started with allowedLatenessMillis " + allowedLatenessMillis);
      ctx.registerEventTimeTimer(System.currentTimeMillis() + allowedLatenessMillis);
    }

    return TriggerResult.CONTINUE;
  }

  @Override
  public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
    log.info("onEvenTime called at "+System.currentTimeMillis() );
    return TriggerResult.FIRE_AND_PURGE;
  }


  @Override
  public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
    return TriggerResult.CONTINUE;
  }

  @Override
  public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {}
}

【问题讨论】：

标签： java apache-flink amazon-kinesis amazon-kinesis-analytics

【解决方案1】：

老实说，我看不出在这里使用GlobalWindow 的理由。您可以只使用与您的Trigger 具有相同目的的KeyedProcessFunction，基本上，它将从事件开始到事件结束的所有元素收集到ListState，然后当您收到isEventEnd()==true，您可以简单地安排 EventTime 计时器，该计时器将在一分钟后触发并发出在 ListState 中收集的结果。

【讨论】：

感谢您的回复。但是，我将如何清除窗口数据？就我而言，传感器停止信号意味着传感器停止在当前间隔内发送数据。我需要应用聚合逻辑，然后将数据保存到数据库。一旦传感器再次打开，它会发送一组具有相同id 的新数据。在那个窗口中，我不想要过去的传感器数据。这就是选择 Global Window 的原因。
是的，但没有什么能阻止您在元素发出后从给定的ListState 中删除所有元素 :) 因为它是键控状态，这意味着您基本上会为此清除所有数据特定的键。

【解决方案2】：

最后，我能够使用以下自定义触发器实现我的要求。

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.windowing.time.Time;
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.GlobalWindow;

@Slf4j
public class JourneyTrigger extends Trigger<SensorEvent, GlobalWindow> {

  private final long allowedLatenessMillis;

  public JourneyTrigger(Time allowedLateness) {
    this.allowedLatenessMillis = allowedLateness.toMilliseconds();
  }

  @Override
  public TriggerResult onElement(SensorEvent element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {

    if (element.isEventEnd()==true) {
      log.info("Timer started with allowedLatenessMillis " + allowedLatenessMillis);
      ctx.registerProcessingTimeTimer(System.currentTimeMillis() + allowedLatenessMillis);
    }

    return TriggerResult.CONTINUE;
  }

  @Override
  public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
    log.info("onProcessingTime called at "+System.currentTimeMillis() );
    return TriggerResult.FIRE_AND_PURGE;
  }

  @Override
  public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
    return TriggerResult.CONTINUE;
  }



  @Override
  public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {}
}

同样在Driver.java类中，设置环境时间特性

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

【讨论】：