Flink – Trigger，Evictor

public abstract class Trigger<T, W extends Window> implements Serializable { /**
* Called for every element that gets added to a pane. The result of this will determine
* whether the pane is evaluated to emit results.
*
* @param element The element that arrived.
* @param timestamp The timestamp of the element that arrived.
* @param window The window to which the element is being added.
* @param ctx A context object that can be used to register timer callbacks.
*/
public abstract TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception; /**
* Called when a processing-time timer that was set using the trigger context fires.
*
* @param time The timestamp at which the timer fired.
* @param window The window for which the timer fired.
* @param ctx A context object that can be used to register timer callbacks.
*/
public abstract TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception; /**
* Called when an event-time timer that was set using the trigger context fires.
*
* @param time The timestamp at which the timer fired.
* @param window The window for which the timer fired.
* @param ctx A context object that can be used to register timer callbacks.
*/
public abstract TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception; /**
* Called when several windows have been merged into one window by the
* {@link org.apache.flink.streaming.api.windowing.assigners.WindowAssigner}.
*
* @param window The new window that results from the merge.
* @param ctx A context object that can be used to register timer callbacks and access state.
*/
public TriggerResult onMerge(W window, OnMergeContext ctx) throws Exception {
throw new RuntimeException("This trigger does not support merging.");
}

/**
* A {@link Trigger} that fires once the current system time passes the end of the window
* to which a pane belongs.
*/
public class ProcessingTimeTrigger implements Trigger<Object, TimeWindow> {
private static final long serialVersionUID = 1L; private ProcessingTimeTrigger() {} @Override
public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {
ctx.registerProcessingTimeTimer(window.maxTimestamp()); //对于processingTime，element的trigger时间是current+window，所以这里需要注册定时器去触发
return TriggerResult.CONTINUE;
} @Override
public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
return TriggerResult.CONTINUE;
} @Override
public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {//触发后调用
return TriggerResult.FIRE_AND_PURGE;
} @Override
public String toString() {
return "ProcessingTimeTrigger()";
} /**
* Creates a new trigger that fires once system time passes the end of the window.
*/
public static ProcessingTimeTrigger create() {
return new ProcessingTimeTrigger();
}
}

public class CountTrigger<W extends Window> extends Trigger<Object, W> { private final long maxCount; private final ReducingStateDescriptor<Long> stateDesc =
new ReducingStateDescriptor<>("count", new Sum(), LongSerializer.INSTANCE); private CountTrigger(long maxCount) {
this.maxCount = maxCount;
} @Override
public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {
ReducingState<Long> count = ctx.getPartitionedState(stateDesc); //从backend取出conunt state
count.add(1L); //加1
if (count.get() >= maxCount) {
count.clear();
return TriggerResult.FIRE;
}
return TriggerResult.CONTINUE;
} @Override
public TriggerResult onEventTime(long time, W window, TriggerContext ctx) {
return TriggerResult.CONTINUE;
} @Override
public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {
return TriggerResult.CONTINUE;
} @Override
public TriggerResult onMerge(W window, OnMergeContext ctx) throws Exception {
ctx.mergePartitionedState(stateDesc); //先调用merge，底层backend里面的window进行merge
ReducingState<Long> count = ctx.getPartitionedState(stateDesc); //merge后再取出state，count，进行判断
if (count.get() >= maxCount) {
return TriggerResult.FIRE;
}
return TriggerResult.CONTINUE;
}

enum TriggerResult {
CONTINUE(false, false), FIRE_AND_PURGE(true, true), FIRE(true, false), PURGE(false, true); private final boolean fire;
private final boolean purge;
}

/**
* Merges two {@code TriggerResults}. This specifies what should happen if we have
* two results from a Trigger, for example as a result from
* {@link Trigger#onElement(Object, long, Window, Trigger.TriggerContext)} and
* {@link Trigger#onEventTime(long, Window, Trigger.TriggerContext)}.
*
* <p>
* For example, if one result says {@code CONTINUE} while the other says {@code FIRE}
* then {@code FIRE} is the combined result;
*/
public static TriggerResult merge(TriggerResult a, TriggerResult b) {
if (a.purge || b.purge) {
if (a.fire || b.fire) {
return FIRE_AND_PURGE;
} else {
return PURGE;
}
} else if (a.fire || b.fire) {
return FIRE;
} else {
return CONTINUE;
}
}

/**
* A context object that is given to {@link Trigger} methods to allow them to register timer
* callbacks and deal with state.
*/
public interface TriggerContext { long getCurrentProcessingTime();
long getCurrentWatermark(); /**
* Register a system time callback. When the current system time passes the specified
* time {@link Trigger#onProcessingTime(long, Window, TriggerContext)} is called with the time specified here.
*
* @param time The time at which to invoke {@link Trigger#onProcessingTime(long, Window, TriggerContext)}
*/
void registerProcessingTimeTimer(long time);
void registerEventTimeTimer(long time); void deleteProcessingTimeTimer(long time);
void deleteEventTimeTimer(long time); <S extends State> S getPartitionedState(StateDescriptor<S, ?> stateDescriptor);
}

public interface OnMergeContext extends TriggerContext {
<S extends MergingState<?, ?>> void mergePartitionedState(StateDescriptor<S, ?> stateDescriptor);
}

/**
* {@code Context} is a utility for handling {@code Trigger} invocations. It can be reused
* by setting the {@code key} and {@code window} fields. No internal state must be kept in
* the {@code Context}
*/
public class Context implements Trigger.OnMergeContext {
protected K key; //Context对应的window上下文
protected W window; protected Collection<W> mergedWindows; //onMerge中被赋值 @SuppressWarnings("unchecked")
public <S extends State> S getPartitionedState(StateDescriptor<S, ?> stateDescriptor) {
try {
return WindowOperator.this.getPartitionedState(window, windowSerializer, stateDescriptor); //从backend里面读出改window的状态，即window buffer
} catch (Exception e) {
throw new RuntimeException("Could not retrieve state", e);
}
} @Override
public <S extends MergingState<?, ?>> void mergePartitionedState(StateDescriptor<S, ?> stateDescriptor) {
if (mergedWindows != null && mergedWindows.size() > 0) {
try {
WindowOperator.this.getStateBackend().mergePartitionedStates(window, //在backend层面把mergedWindows merge到window中
mergedWindows,
windowSerializer,
stateDescriptor);
} catch (Exception e) {
throw new RuntimeException("Error while merging state.", e);
}
}
} @Override
public void registerProcessingTimeTimer(long time) {
Timer<K, W> timer = new Timer<>(time, key, window);
// make sure we only put one timer per key into the queue
if (processingTimeTimers.add(timer)) {
processingTimeTimersQueue.add(timer);
//If this is the first timer added for this timestamp register a TriggerTask
if (processingTimeTimerTimestamps.add(time, 1) == 0) { //如果这个window是第一次注册的话
ScheduledFuture<?> scheduledFuture = WindowOperator.this.registerTimer(time, WindowOperator.this); //对于processTime必须注册定时器主动触发
processingTimeTimerFutures.put(time, scheduledFuture);
}
}
} @Override
public void registerEventTimeTimer(long time) {
Timer<K, W> timer = new Timer<>(time, key, window);
if (watermarkTimers.add(timer)) {
watermarkTimersQueue.add(timer);
}
} //封装一遍trigger的接口，并把self作为context传入trigger的接口中
public TriggerResult onElement(StreamRecord<IN> element) throws Exception {
return trigger.onElement(element.getValue(), element.getTimestamp(), window, this);
} public TriggerResult onProcessingTime(long time) throws Exception {
return trigger.onProcessingTime(time, window, this);
} public TriggerResult onEventTime(long time) throws Exception {
return trigger.onEventTime(time, window, this);
} public TriggerResult onMerge(Collection<W> mergedWindows) throws Exception {
this.mergedWindows = mergedWindows;
return trigger.onMerge(window, this);
}}

/**
* An {@code Evictor} can remove elements from a pane before it is being processed and after
* window evaluation was triggered by a
* {@link org.apache.flink.streaming.api.windowing.triggers.Trigger}.
*
* <p>
* A pane is the bucket of elements that have the same key (assigned by the
* {@link org.apache.flink.api.java.functions.KeySelector}) and same {@link Window}. An element can
* be in multiple panes of it was assigned to multiple windows by the
* {@link org.apache.flink.streaming.api.windowing.assigners.WindowAssigner}. These panes all
* have their own instance of the {@code Evictor}.
*
* @param <T> The type of elements that this {@code Evictor} can evict.
* @param <W> The type of {@link Window Windows} on which this {@code Evictor} can operate.
*/
public interface Evictor<T, W extends Window> extends Serializable { /**
* Computes how many elements should be removed from the pane. The result specifies how
* many elements should be removed from the beginning.
*
* @param elements The elements currently in the pane.
* @param size The current number of elements in the pane.
* @param window The {@link Window}
*/
int evict(Iterable<StreamRecord<T>> elements, int size, W window);
}

/**
* An {@link Evictor} that keeps only a certain amount of elements.
*
* @param <W> The type of {@link Window Windows} on which this {@code Evictor} can operate.
*/
public class CountEvictor<W extends Window> implements Evictor<Object, W> {
private static final long serialVersionUID = 1L; private final long maxCount; private CountEvictor(long count) {
this.maxCount = count;
} @Override
public int evict(Iterable<StreamRecord<Object>> elements, int size, W window) {
if (size > maxCount) {
return (int) (size - maxCount);
} else {
return 0;
}
} /**
* Creates a {@code CountEvictor} that keeps the given number of elements.
*
* @param maxCount The number of elements to keep in the pane.
*/
public static <W extends Window> CountEvictor<W> of(long maxCount) {
return new CountEvictor<>(maxCount);
}
}

/**
* An {@link Evictor} that keeps elements for a certain amount of time. Elements older
* than {@code current_time - keep_time} are evicted.
*
* @param <W> The type of {@link Window Windows} on which this {@code Evictor} can operate.
*/
public class TimeEvictor<W extends Window> implements Evictor<Object, W> {
private static final long serialVersionUID = 1L; private final long windowSize; public TimeEvictor(long windowSize) {
this.windowSize = windowSize;
} @Override
public int evict(Iterable<StreamRecord<Object>> elements, int size, W window) {
int toEvict = 0;
long currentTime = Iterables.getLast(elements).getTimestamp();
long evictCutoff = currentTime - windowSize;
for (StreamRecord<Object> record: elements) {
if (record.getTimestamp() > evictCutoff) {
break;
}
toEvict++;
}
return toEvict;
}
}