Flink 源码分析之写给大忙人看的 Flink Window 原理

package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.slf4j.LoggerFactory;
import java.util.Properties;
/**
 * @author shengjk1
 * @date 2019/9/4
 */
public class Main {
	protected final static org.slf4j.Logger logger = LoggerFactory.getLogger(Main.class);
	
	public static void main(String[] args) throws Exception {
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		
		env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
		
		env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);
		env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
		env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
		env.getCheckpointConfig().setCheckpointTimeout(60000);
		env.getCheckpointConfig().setMaxConcurrentCheckpoints(5);
		env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
		env.getCheckpointConfig().setFailOnCheckpointingErrors(false);
		
		env.setParallelism(1);
		
		StateBackend backend =
			new RocksDBStateBackend("file:////Users/iss/sourceCode/spark/flink/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/checkpoints", true);
		env.setStateBackend(backend);
		Properties properties = new Properties();
		properties.setProperty("bootstrap.servers", "bigdata-dev-mq:9092");
		properties.setProperty("group.id", "test");
		properties.setProperty(FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS, "1000");
		
		
		FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties);
		consumer.setStartFromEarliest();
		
		env.addSource(consumer).uid("orderAndRegisterUserIdSource")
				.rebalance()
				.keyBy(new KeySelector<String, String>() {
					@Override
					public String getKey(String value) throws Exception {
						return value;
					}
				})
				.timeWindow(Time.seconds(2))
				.trigger(new CountAndTimeTrigger(2L)
				.process(new ProcessWindowFunctionImp()).uid("process");
		
		
		// execute program
		env.execute("realTimeDataWareHouse");
	}
}

// window operator 的 processElement
	public void processElement(StreamRecord<IN> element) throws Exception {
		final Collection<W> elementWindows = windowAssigner.assignWindows(
			element.getValue(), element.getTimestamp(), windowAssignerContext);
		//if element is handled by none of assigned elementWindows
		boolean isSkippedElement = true;
		final K key = this.<K>getKeyedStateBackend().getCurrentKey();
		if (windowAssigner instanceof MergingWindowAssigner) {
			MergingWindowSet<W> mergingWindows = getMergingWindowSet();
			for (W window: elementWindows) {
				// adding the new window might result in a merge, in that case the actualWindow
				// is the merged window and we work with that. If we don't merge then
				// actualWindow == window
				W actualWindow = mergingWindows.addWindow(window, new MergingWindowSet.MergeFunction<W>() {
					@Override
					public void merge(W mergeResult,
							Collection<W> mergedWindows, W stateWindowResult,
							Collection<W> mergedStateWindows) throws Exception {
						if ((windowAssigner.isEventTime() && mergeResult.maxTimestamp() + allowedLateness <= internalTimerService.currentWatermark())) {
							throw new UnsupportedOperationException("The end timestamp of an " +
									"event-time window cannot become earlier than the current watermark " +
									"by merging. Current watermark: " + internalTimerService.currentWatermark() +
									" window: " + mergeResult);
						} else if (!windowAssigner.isEventTime() && mergeResult.maxTimestamp() <= internalTimerService.currentProcessingTime()) {
							throw new UnsupportedOperationException("The end timestamp of a " +
									"processing-time window cannot become earlier than the current processing time " +
									"by merging. Current processing time: " + internalTimerService.currentProcessingTime() +
									" window: " + mergeResult);
						}
						triggerContext.key = key;
						triggerContext.window = mergeResult;
						triggerContext.onMerge(mergedWindows);
						for (W m: mergedWindows) {
							triggerContext.window = m;
							triggerContext.clear();
							deleteCleanupTimer(m);
						}
						// merge the merged state windows into the newly resulting state window
						windowMergingState.mergeNamespaces(stateWindowResult, mergedStateWindows);
					}
				});
				// drop if the window is already late
				if (isWindowLate(actualWindow)) {
					mergingWindows.retireWindow(actualWindow);
					continue;
				}
				isSkippedElement = false;
				W stateWindow = mergingWindows.getStateWindow(actualWindow);
				if (stateWindow == null) {
					throw new IllegalStateException("Window " + window + " is not in in-flight window set.");
				}
				windowState.setCurrentNamespace(stateWindow);
				windowState.add(element.getValue());
				triggerContext.key = key;
				triggerContext.window = actualWindow;
				TriggerResult triggerResult = triggerContext.onElement(element);
				if (triggerResult.isFire()) {
					// RockdbListState  RocksDBReducingState
					ACC contents = windowState.get();
					if (contents == null) {
						continue;
					}
					emitWindowContents(actualWindow, contents);
				}
				if (triggerResult.isPurge()) {
					windowState.clear();
				}
				registerCleanupTimer(actualWindow);
			}
			// need to make sure to update the merging state in state
			mergingWindows.persist();
		} else {
			for (W window: elementWindows) {
				// drop if the window is already late
				if (isWindowLate(window)) {
					continue;
				}
				isSkippedElement = false;
				windowState.setCurrentNamespace(window);
				//数据过来之后会先存入 windowState 直至 window fire
				windowState.add(element.getValue());
				triggerContext.key = key;
				triggerContext.window = window;
				//调用用户定义的 onElement 代码
				TriggerResult triggerResult = triggerContext.onElement(element);
				//当触发窗口时，从 windowState 中获取数据，在本样例中 windowState 为 RocksDBListState
				if (triggerResult.isFire()) {
					//RocksDBListState RocksDBReducingState
					//
					ACC contents = windowState.get();
					if (contents == null) {
						continue;
					}
					//当窗口触发时，会将 window 中数据发送到下游,调用用户的 process 方法。
					emitWindowContents(window, contents);
				}
				if (triggerResult.isPurge()) {
					windowState.clear();
				}
				// 注册 timer，其实就是定时调度任务。底层通过 ScheduledThreadPoolExecutor.schedule(...)来实现的
				// 每个窗口中的每个 key 会有且仅有一个 timer( 判断方式的一部分是通过 map 来实现的)
				registerCleanupTimer(window);
			}
		}

// 这个是通过 timer 来调用的，
	// processElement 的时候 registerCleanupTimer(window) 会创建相应的 timer
	public void onProcessingTime(InternalTimer<K, W> timer) throws Exception {
		triggerContext.key = timer.getKey();
		triggerContext.window = timer.getNamespace();
		MergingWindowSet<W> mergingWindows;
		if (windowAssigner instanceof MergingWindowAssigner) {
			mergingWindows = getMergingWindowSet();
			W stateWindow = mergingWindows.getStateWindow(triggerContext.window);
			if (stateWindow == null) {
				// Timer firing for non-existent window, this can only happen if a
				// trigger did not clean up timers. We have already cleared the merging
				// window and therefore the Trigger state, however, so nothing to do.
				return;
			} else {
				windowState.setCurrentNamespace(stateWindow);
			}
		} else {
			windowState.setCurrentNamespace(triggerContext.window);
			mergingWindows = null;
		}
		TriggerResult triggerResult = triggerContext.onProcessingTime(timer.getTimestamp());
		if (triggerResult.isFire()) {
			ACC contents = windowState.get();
			if (contents != null) {
				emitWindowContents(triggerContext.window, contents);
			}
		}
		if (triggerResult.isPurge()) {
			windowState.clear();
		}
		if (!windowAssigner.isEventTime() && isCleanupTime(triggerContext.window, timer.getTimestamp())) {
			// 会清空所有的 state
			// 先 windowState.clear() 调用用户定义的 clear 方法，然后再清除 windowContext 内部的状态：
			// 仅仅是通过 onProcessingTime or onEventTime method fire window 才可能会触发 clearAllState 操作
			// 否则会可以理解为还是一个窗口虽然 fire 了。
			// 先增量增量的 fire 然后再全量的 fire ( onProcessingTime and  onEventTime  导致的 fire ，未指定 purge)
			clearAllState(triggerContext.window, windowState, mergingWindows);
		}
		if (mergingWindows != null) {
			// need to make sure to update the merging state in state
			mergingWindows.persist();
		}
	}