我已经实现了自定义WindowAssigher
:
public class SessionWindowAssigner extends WindowAssigner<LogItem, SessionWindow> {
@Override
public Collection<SessionWindow> assignWindows(LogItem element, long timestamp) {
return Collections.singletonList(new SessionWindow(element.getSessionUid()));
}
@Override
public Trigger<LogItem, SessionWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
return new SessionTrigger(60_000L);
}
@Override
public TypeSerializer<SessionWindow> getWindowSerializer(ExecutionConfig executionConfig) {
return new SessionWindow.Serializer();
}
}
, Window
:
public class SessionWindow extends Window {
private final String sessionUid;
public SessionWindow(String sessionUid) {
this.sessionUid = sessionUid;
}
public String getSessionUid() {
return sessionUid;
}
@Override
public long maxTimestamp() {
return Long.MAX_VALUE;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
SessionWindow that = (SessionWindow) o;
return sessionUid.equals(that.sessionUid);
}
@Override
public int hashCode() {
return sessionUid.hashCode();
}
public static class Serializer extends TypeSerializer<SessionWindow> {
private static final long serialVersionUID = 1L;
@Override
public boolean isImmutableType() {
return true;
}
@Override
public TypeSerializer<SessionWindow> duplicate() {
return this;
}
@Override
public SessionWindow createInstance() {
return null;
}
@Override
public SessionWindow copy(SessionWindow from) {
return from;
}
@Override
public SessionWindow copy(SessionWindow from, SessionWindow reuse) {
return from;
}
@Override
public int getLength() {
return 0;
}
@Override
public void serialize(SessionWindow record, DataOutputView target) throws IOException {
target.writeUTF(record.sessionUid);
}
@Override
public SessionWindow deserialize(DataInputView source) throws IOException {
return new SessionWindow(source.readUTF());
}
@Override
public SessionWindow deserialize(SessionWindow reuse, DataInputView source) throws IOException {
return new SessionWindow(source.readUTF());
}
@Override
public void copy(DataInputView source, DataOutputView target) throws IOException {
target.writeUTF(source.readUTF());
}
@Override
public boolean equals(Object obj) {
return obj instanceof Serializer;
}
@Override
public boolean canEqual(Object obj) {
return obj instanceof Serializer;
}
@Override
public int hashCode() {
return 0;
}
}
}
和Trigger
:
public class SessionTrigger extends Trigger<LogItem, SessionWindow> {
private final long sessionTimeout;
private final ValueStateDescriptor<Long> previousFinishTimestampDesc = new ValueStateDescriptor<>("SessionTrigger.timestamp", LongSerializer.INSTANCE, null);
public SessionTrigger(long sessionTimeout) {
this.sessionTimeout = sessionTimeout;
}
@Override
public TriggerResult onElement(LogItem element, long timestamp, SessionWindow window, TriggerContext ctx) throws Exception {
ValueState<Long> previousFinishTimestampState = ctx.getPartitionedState(previousFinishTimestampDesc);
Long previousFinishTimestamp = previousFinishTimestampState.value();
Long newFinisTimestamp = timestamp + sessionTimeout;
if (previousFinishTimestamp != null) {
ctx.deleteEventTimeTimer(previousFinishTimestamp);
}
ctx.registerEventTimeTimer(newFinisTimestamp);
previousFinishTimestampState.update(newFinisTimestamp);
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onEventTime(long time, SessionWindow window, TriggerContext ctx) throws Exception {
return TriggerResult.FIRE_AND_PURGE;
}
@Override
public TriggerResult onProcessingTime(long time, SessionWindow window, TriggerContext ctx) throws Exception {
throw new UnsupportedOperationException("This is not processing time trigger");
}
@Override
public void clear(SessionWindow window, TriggerContext ctx) throws Exception {
ValueState<Long> previousFinishTimestampState = ctx.getPartitionedState(previousFinishTimestampDesc);
Long previousFinishTimestamp = previousFinishTimestampState.value();
ctx.deleteEventTimeTimer(previousFinishTimestamp);
previousFinishTimestampState.clear();
}
}
用于通过超时检测会话结束,即如果最后一个事件是 N 秒前,则评估窗口函数。如您所见,我在 ValueState 中保存最后一个事件时间戳,因为我想在失败后恢复它。
似乎我应该Checkpointed
在此触发器中实现保存/恢复保存点(和检查点)快照的接口,因为我不想在重新部署流程期间丢失触发器状态。
那么,任何人都可以向我解释如何SessionTrigger
在部署期间正确保存触发器(以及可能相关的窗口)的状态吗?
据我了解,我只应该实现Checkpointed
接口,SessionTrigger
因为只有它有状态。正确的?SessionWindow
-s 和怎么样SessionWindowAssigner
?它们会在自动部署后恢复还是我应该手动进行?