我为我的事件流创建了一个自定义触发器和处理函数。
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 {}
}