點選上方藍色字型，選擇“設為星标”
回複”資源“擷取更多資源
Flink 的 window 有兩個基本款，TimeWindow 和 CountWindow。
TimeWindow 是到時間就觸發視窗，CountWindow 是到數量就觸發。
如果我需要到時間就觸發，并且到時間之前如果已經積累了足夠數量的資料；或者在限定時間内沒有積累足夠數量的資料，我依然希望觸發視窗業務，那麼就需要自定義觸發器。
import org.apache.flink.api.common.functions.ReduceFunction;import org.apache.flink.api.common.state.ReducingState;import org.apache.flink.api.common.state.ReducingStateDescriptor;import org.apache.flink.api.common.typeutils.base.LongSerializer;import org.apache.flink.streaming.api.TimeCharacteristic;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.TimeWindow;import org.slf4j.Logger;import org.slf4j.LoggerFactory;/** * 帶逾時的計數視窗觸發器 */public class CountTriggerWithTimeout<T> extends Trigger<T, TimeWindow> { private static Logger LOG = LoggerFactory.getLogger(CountTriggerWithTimeout.class); /** * 視窗最大資料量 */ private int maxCount; /** * event time / process time */ private TimeCharacteristic timeType; /** * 用于儲存視窗目前資料量的狀态對象 */ private ReducingStateDescriptor countStateDescriptor = new ReducingStateDescriptor("counter", new Sum(), LongSerializer.INSTANCE); public CountTriggerWithTimeout(int maxCount, TimeCharacteristic timeType) { this.maxCount = maxCount; this.timeType = timeType; } private TriggerResult fireAndPurge(TimeWindow window, TriggerContext ctx) throws Exception { clear(window, ctx); return TriggerResult.FIRE_AND_PURGE; } @Override public TriggerResult onElement(T element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception { ReducingState countState = ctx.getPartitionedState(countStateDescriptor); countState.add(1L); if (countState.get() >= maxCount) { LOG.info("fire with count: " + countState.get()); return fireAndPurge(window, ctx); } if (timestamp >= window.getEnd()) { LOG.info("fire with tiem: " + timestamp); return fireAndPurge(window, ctx); } else { return TriggerResult.CONTINUE; } } @Override public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception { if (timeType != TimeCharacteristic.ProcessingTime) { return TriggerResult.CONTINUE; } if (time >= window.getEnd()) { return TriggerResult.CONTINUE; } else { LOG.info("fire with process tiem: " + time); return fireAndPurge(window, ctx); } } @Override public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception { if (timeType != TimeCharacteristic.EventTime) { return TriggerResult.CONTINUE; } if (time >= window.getEnd()) { return TriggerResult.CONTINUE; } else { LOG.info("fire with event tiem: " + time); return fireAndPurge(window, ctx); } } @Override public void clear(TimeWindow window, TriggerContext ctx) throws Exception { ReducingState countState = ctx.getPartitionedState(countStateDescriptor); countState.clear(); } /** * 計數方法 */ class Sum implements ReduceFunction<Long> { @Override public Long reduce(Long value1, Long value2) throws Exception { return value1 + value2; } }}
使用示例(逾時時間 10 秒，資料量上限 1000)：
stream .timeWindowAll(Time.seconds(10)) .trigger( new CountTriggerWithTimeout(1000, TimeCharacteristic.ProcessingTime) ) .process(new XxxxWindowProcessFunction()) .addSink(new XxxSinkFunction()) .name("Xxx");
即可。
歡迎點贊+收藏+轉發朋友圈素質三連
文章不錯？點個【在看】吧！ 👇