Flink SQL Window源碼全解析

文章目錄

一、概述

二、Window分類

1、TimeWindow與CountWindow

2、TimeWindow子類型

• Tumble Window（翻轉窗口）
• Hop Window（滑動窗口）
• Session Window（會話窗口）

三、Window分類及整體流程

四、創建WindowOperator算子

五、WindowOperator處理數據圖解

六、WindowOperator源碼調試

1、StreamExecGroupWindowAggregate#createWindowOperator()創建算子

2、WindowOperator#processElement()處理數據，注冊Timer

3、Timer觸發

• InternalTimerServiceImpl#advanceWatermark()
• WindwOperator#onEventTime()
• emitWindowResult()提交結果

七、Emit（Trigger）觸發器

• 1、Emit策略
• 2、用途
• 3、語法
• 4、示例
• 5、Trigger類和結構關系

概述

窗口是無限流上一種核心機制，可以流分割為有限大小的“窗口”，同時，在窗口內進行聚合，從而把源源不斷產生的數據根據不同的條件划分成一段一段有邊界的數據區間，使用戶能夠利用窗口功能實現很多復雜的統計分析需求。

本文內容：

• Flink SQL WINDOW功能介紹
• 底層實現源碼分析：StreamExecGroupWindowAggregate創建WindowOperator
• 底層實現源碼分析：WindowOperator算子處理數據這兩個地方源碼分析。

Window分類

1、TimeWindow與CountWindow
Flink Window可以是時間驅動的（TimeWindow），也可以是數據驅動的（CountWindow）。
由於flink-planner-blink SQL中目前只支持TimeWindow相應的表達語句（TUMBLE、HOP、SESSION），因此，本文主要介紹TimeWindow SQL示例和邏輯，CountWindow感興趣的讀者可自行分析。

2、TimeWindow子類型
Flink TimeWindow有滑動窗口(HOP)、滾動窗口(TUMBLE)以及會話窗口(SESSION)三種，所選取的字段時間，可以是系統時間(PROCTIME)或事件時間(EVENT TIME)兩種，接來下依次介紹。

• Tumble Window（翻轉窗口）

翻轉窗口Assigner將每個元素分配給具有指定大小的窗口。翻轉窗口的大小是固定的，且不會重疊。例如，指定一個大小為5分鍾的翻滾窗口，並每5分鍾啟動一個新窗口，如下圖所示：

TUMBLE ROWTIME語法示例：

CREATE TABLE sessionOrderTableRowtime (
	ctime TIMESTAMP,
	categoryName VARCHAR,
	shopName VARCHAR,
	itemName VARCHAR,
	userId VARCHAR,
	price FLOAT,
	action BIGINT,
	WATERMARK FOR ctime AS withOffset(ctime, 1000),
	proc AS PROCTIME()
) with (
	`type` = 'kafka',
	format = 'json',
	updateMode = 'append',
	`group.id` = 'groupId',
	bootstrap.servers = 'xxxxx:9092',
	version = '0.10',
	`zookeeper.connect` = 'xxxxx:2181',
	startingOffsets = 'latest',
	topic = 'sessionsourceproctime'
);


CREATE TABLE popwindowsink (
	countA BIGINT,
	ctime_start TIMESTAMP,
	ctime_end VARCHAR,
	ctime_rowtime VARCHAR,
	categoryName VARCHAR,
	price_sum FLOAT
) with (
	format = 'json',
	updateMode = 'append',
	bootstrap.servers = 'xxxxx:9092',
	version = '0.10',
	topic = 'sessionsinkproctime',
	`type` = 'kafka'
);

INSERT INTO popwindowsink
(SELECT
COUNT(*),
TUMBLE_START(ctime, INTERVAL '5' MINUTE),
DATE_FORMAT(TUMBLE_END(ctime, INTERVAL '5' MINUTE), 'yyyy-MM-dd-HH-mm-ss:SSS'), --將TUMBLE_END轉為可視化的日期
DATE_FORMAT(TUMBLE_ROWTIME(ctime, INTERVAL '5' MINUTE), 'yyyy-MM-dd-HH-mm-ss:SSS'), --這里TUMBLE_ROWTIME為TUMBLE_END-1ms，一般用於后續窗口級聯聚合
categoryName,
SUM(price)
FROM sessionOrderTableRowtime
GROUP BY TUMBLE(ctime, INTERVAL '5' MINUTE), categoryName)

TUMBLEP ROCTIME語法示例：

INSERT INTO popwindowsink
(SELECT
COUNT(*),
TUMBLE_START(proc, INTERVAL '5' MINUTE),
DATE_FORMAT(TUMBLE_END(proc, INTERVAL '5' MINUTE), 'yyyy-MM-dd-HH-mm-ss:SSS'),
DATE_FORMAT(TUMBLE_PROCTIME(proc, INTERVAL '5' MINUTE), 'yyyy-MM-dd-HH-mm-ss:SSS'), --注意這里proc字段即Source DDL中指定的PROCTIME
categoryName,
SUM(price)
FROM sessionOrderTableRowtime
GROUP BY TUMBLE(proc, INTERVAL '5' MINUTE), categoryName)

ROWTIME與PROCTIME區別：

• 在使用上： 主要是填入的ctime、proc關鍵字的區別，這兩個字段在Source DDL中指定方式不一樣.
• 在實現原理上： ROWTIME模式，根據ctime對應的值，去確定窗口的start、end；PROCTIME模式，在WindowOperator處理數據時，獲取本地系統時間，去確定窗口的start、end.

由於生產系統中，主要使用ROWTIME來計算、聚合、統計，PROCTIME一般用於測試或對統計精度要求不高的場景，本文后續都主要以ROWTIME進行分析。

• Hop Window（滑動窗口）

滑動窗口Assigner將元素分配給多個固定長度的窗口。類似於滾動窗口分配程序，窗口的大小由窗口大小參數配置。因此，如果滑動窗口小於窗口大小，則滑動窗口可以重疊。在這種情況下，元素被分配到多個窗口。其實，滾動窗口TUMBLE是滑動窗口的一個特例。
例子，設置一個10分鍾長度的窗口，以5分鍾間隔滑動。這樣，每5分鍾就會出現一個窗口，其中包含最近10分鍾內到達的事件，如下圖：

HOP ROWTIME語法示例：

INSERT INTO popwindowsink
(SELECT
COUNT(*),
HOP_START(ctime, INTERVAL '5' MINUTE,  INTERVAL '10' MINUTE),
DATE_FORMAT(HOP_END(ctime, INTERVAL '5' MINUTE,  INTERVAL '10' MINUTE), 'yyyy-MM-dd-HH-mm-ss:SSS'),
DATE_FORMAT(HOP_ROWTIME(ctime, INTERVAL '5' MINUTE,  INTERVAL '10' MINUTE), 'yyyy-MM-dd-HH-mm-ss:SSS'), --注意這里ctime字段即Source DDL中指定的ROWTIME
categoryName,
SUM(price)
FROM sessionOrderTableRowtime
GROUP BY HOP(ctime, INTERVAL '5' MINUTE,  INTERVAL '10' MINUTE), categoryName)

• Session Window（會話窗口）

會話窗口Assigner根據活動會話對元素進行分組。與翻滾窗口和滑動窗口相比，會話窗口不會重疊，也沒有固定的開始和結束時間。相反，會話窗口在一段時間內不接收元素時關閉，即，當一段不活躍的間隙發生時，當前會話關閉，隨后的元素被分配給新的會話。

SESSION ROWTIME語法示例：

INSERT INTO popwindowsink
(SELECT
COUNT(*),
SESSION_START(ctime, INTERVAL '5' MINUTE),
DATE_FORMAT(SESSION_END(ctime, INTERVAL '5' MINUTE, 'yyyy-MM-dd-HH-mm-ss:SSS'),
DATE_FORMAT(SESSION_ROWTIME(ctime, INTERVAL '5' MINUTE), 'yyyy-MM-dd-HH-mm-ss:SSS'), --注意這里ctime字段即Source DDL中指定的ROWTIME
categoryName,
SUM(price)
FROM sessionOrderTableRowtime
GROUP BY SESSION(ctime, INTERVAL '5' MINUTE), categoryName)

Window分類及整體流程

上圖內部流程分析：

應用層SQL:
1.1 window分類及配置，包括滑動、翻轉、會話類型窗口
1.2 window時間類型配置，默認待字段名的EventTime，也可以通過PROCTIME()配置為ProcessingTime
Calcite解析引擎:
2.1 Calcite SQL解析，包括邏輯、優化、物理計划和算子綁定(#translateToPlanInternal)，在本文特指StreamExecGroupWindowAggregateRule和StreamExecGroupWindowAggregate物理計划
WindowOperator算子創建相關:
3.1 StreamExecGroupWindowAggregate#createWindowOperator創建算子
3.2 WindowAssigner的創建，根據輸入的數據，和窗口類型，生成多個窗口
3.3 processElement()真實處理數據，包括聚合運算，生成窗口，更新緩存，提交數據等功能
3.4 Trigger根據數據或時間，來決定窗口觸發

創建WindowOperator算子

由於window語法主要是在group by語句中使用，calcite創建WindowOperator算子伴隨着聚合策略的實現，包括聚合規則匹配(StreamExecGroupWindowAggregateRule)，以及生成聚合physical算子StreamExecGroupWindowAggregate兩個子流程：

上圖內部流程分析：

a. StreamExecGroupWindowAggregateRule會對window進行提前匹配，
生成的WindowEmitStrategy內部具有：是否為EventTime表標識、是否為SessionWindow、early fire和late fire配置、延遲毫秒數（窗口結束時間加上這個毫秒數即數據清理時間）
b. StreamExecGroupWindowAggregateRule會獲取聚合邏輯計划中，window配置的時間字段，記錄時間字段index信息，window的觸發和清理都會用到這個時間
c. StreamExecGroupWindowAggregate入口即為translateToPlanInternal，它的實現方式與spark比較類似，會先循環調用child子節點translateToPlan方法，生成inputtranform信息作為輸入
d.創建aggregateHandler是一個代碼生成的過程，其生成的創建的class實現了accumulate、retract、merge、update方法，這個handler最后也傳遞給了WindowOperater，處理數據時，可以進行聚合、回撤並輸出最新數據給下游
e. StreamExecGroupWindowAggregate與window相關的最后一步就是調用#createWindowOperator創建算子，其內部先創建了一個WindowOperatorBuilder，設置window類型、retract標識、trigger(window觸發條件)、聚合函數句柄等，最后創建WindowOperator

WindowOperator處理數據圖解

在上一小節，已經完成了WindowOperator參數的設定，並創建實例，接下來我們主要分析WindowOperator真實處理數據的流程(起點在WindowOperator#processElement方法)：

processElement處理數據流程：

a、 獲取當前record具有的事件時間，如果是Processing Time模式，從時間服務Service里面獲取時間即可
b、使用上一步獲取的時間，接着調用windowFunction.assignWindow生成窗口，其內部實際上是調用各類型的WindowAssigner生成窗口，windowFunction有三大類，分別是Paned（滑動）、Merge（會話）、General（前兩種以外的），WindowAssigner類型大致有5類，分別是Tumbling（翻轉）、Sliding（滑動）、Session（會話）、CountTumbling 、CountSlide這幾類,根據輸入的一條數據和時間，可以生成1到多個窗口
c、接下來是遍歷涉及的窗口進行聚合，包括從windowState獲取聚合前值、使用句柄進行聚合、更新狀態至windowState，將當前轉態
d、上一步聚合完成后，就可以遍歷窗口，使用TriggerContext（其實就是不同類型窗口Trigger觸發器的代理），綜合early fire、late fire、水印時間與窗口結束時間，綜合判斷是否觸發窗口寫出
e、如果TriggerContext判斷出觸發條件為true，則調用emitWindowResult寫出，其內部有retract判斷，更新當前state及previous state，寫出數據等操作
f、如果TriggerContext判斷出觸發條件為false，則觸發需要注冊cleanupTimer,到達指定時間后，觸發onEventTime或onProcessingTime
g、onEventTime或onProcessingTime功能十分類似，首先會觸發emitWindowResult提交結果，另外會判斷窗口結束時間+Lateness和當前時間是否相等，相等則表示可以清除窗口數據、當前state及previous state、窗口對應trigger。

WindowOperator源碼調試

為了更直觀的理解Window內部運行原理，這里我們引入一個Flink源碼中已有的SQL Window測試用例，並進行了簡單的修改（即修改為使用HOP滑動窗口）

class WindowJoinITCase{
  @Test
  def testRowTimeInnerJoinWithWindowAggregateOnFirstTime(): Unit = {
    val sqlQuery =
      """
        |SELECT t1.key, HOP_END(t1.rowtime, INTERVAL '4' SECOND, INTERVAL '20' SECOND), COUNT(t1.key)
        |FROM T1 AS t1
        |GROUP BY HOP(t1.rowtime, INTERVAL '4' SECOND, INTERVAL '20' SECOND), t1.key
        |""".stripMargin

    val data1 = new mutable.MutableList[(String, String, Long)]
    data1.+=(("A", "L-1", 1000L))
    data1.+=(("A", "L-2", 2000L))
    data1.+=(("A", "L-3", 3000L))
    //data1.+=(("B", "L-8", 2000L))
    data1.+=(("B", "L-4", 4000L)) 
    data1.+=(("C", "L-5", 2100L))
    data1.+=(("A", "L-6", 10000L)) 
    data1.+=(("A", "L-7", 13000L))

    val t1 = env.fromCollection(data1)
      .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
      .toTable(tEnv, 'key, 'id, 'rowtime)

    tEnv.registerTable("T1", t1)

    val sink = new TestingAppendSink
    val t_r = tEnv.sqlQuery(sqlQuery)
    val result = t_r.toAppendStream[Row]
    result.addSink(sink)
    env.execute()
  }
}

1、StreamExecGroupWindowAggregate#createWindowOperator()創建算子

StreamExecGroupWindowAggregate#createWindowOperator()是創建WindowOperator算子的地方，對應的代碼和注釋：

class StreamExecGroupWindowAggregate{
  private def createWindowOperator(
      config: TableConfig,
      aggsHandler: GeneratedNamespaceAggsHandleFunction[_],
      recordEqualiser: GeneratedRecordEqualiser,
      accTypes: Array[LogicalType],
      windowPropertyTypes: Array[LogicalType],
      aggValueTypes: Array[LogicalType],
      inputFields: Seq[LogicalType],
      timeIdx: Int): WindowOperator[_, _] = {

    val builder = WindowOperatorBuilder
      .builder()
      .withInputFields(inputFields.toArray)
    val timeZoneOffset = -config.getTimeZone.getOffset(Calendar.ZONE_OFFSET)
	
	// 設置WindowOperatorBuilder，最后通過Builder創建WindowOperator
    val newBuilder = window match {
      case TumblingGroupWindow(_, timeField, size) //Tumble PROCTIME模式，內部設置Assiger
          if isProctimeAttribute(timeField) && hasTimeIntervalType(size) =>
        builder.tumble(toDuration(size), timeZoneOffset).withProcessingTime()

      case TumblingGroupWindow(_, timeField, size) //Tumble ROWTIME模式，內部設置Assiger
          if isRowtimeAttribute(timeField) && hasTimeIntervalType(size) =>
        builder.tumble(toDuration(size), timeZoneOffset).withEventTime(timeIdx)

      case SlidingGroupWindow(_, timeField, size, slide) //HOP PROCTIME模式，內部設置Assiger
          if isProctimeAttribute(timeField) && hasTimeIntervalType(size) =>
        builder.sliding(toDuration(size), toDuration(slide), timeZoneOffset)
          .withProcessingTime()
       .....
      case SessionGroupWindow(_, timeField, gap)
          if isRowtimeAttribute(timeField) =>
        builder.session(toDuration(gap)).withEventTime(timeIdx)
    }

	// Retraction和Trigger設置
	//默認是no retract和EventTime.afterEndOfWindow
    if (emitStrategy.produceUpdates) {
      // mark this operator will send retraction and set new trigger
      newBuilder
        .withSendRetraction()
        .triggering(emitStrategy.getTrigger)
    }

    newBuilder
      .aggregate(aggsHandler, recordEqualiser, accTypes, aggValueTypes, windowPropertyTypes)
      .withAllowedLateness(Duration.ofMillis(emitStrategy.getAllowLateness))
      .build()
  }
}

2、WindowOperator#processElement()處理數據，注冊Timer

public class WindowOperator{
	public void processElement(StreamRecord<BaseRow> record) throws Exception {
		BaseRow inputRow = record.getValue();
		long timestamp;
		// 獲取時間戳（數據時間或系統時間），這個時間是后續邏輯划分窗口的依據
		// 例如獲取的timestamp為10000L
		if (windowAssigner.isEventTime()) {
			timestamp = inputRow.getLong(rowtimeIndex);
		} else {
			timestamp = internalTimerService.currentProcessingTime();
		}

		// 計算當前數據所屬於的窗口，注意滑動窗口這里計算出來也只有一個affected窗口（見調試數據），在這個窗口內進行聚合
		Collection<W> affectedWindows = windowFunction.assignStateNamespace(inputRow, timestamp);
		boolean isElementDropped = true;
		for (W window : affectedWindows) {
			isElementDropped = false;
			// 設置ValueState命名空間，例如TimeWindow{start=8000, end=12000}
			windowState.setCurrentNamespace(window);
			// 從windowState獲取上次聚合值
			BaseRow acc = windowState.value();
			if (acc == null) {
				acc = windowAggregator.createAccumulators();
			}
			windowAggregator.setAccumulators(window, acc);
			// 默認進行聚合
			if (BaseRowUtil.isAccumulateMsg(inputRow)) {
				windowAggregator.accumulate(inputRow);
			} else {
				windowAggregator.retract(inputRow);
			}
			acc = windowAggregator.getAccumulators();
			// 更新TimeWindow{start=8000, end=12000}對應聚合值
			windowState.update(acc);
		}

		// 對應的實際窗口，例如輸入Timestamp為10000L，且執行HOP(t1.rowtime, INTERVAL '4' SECOND, INTERVAL '20' SECOND)，拆分出實際的窗口為：
		// TimeWindow{start=-8000, end=12000}
		// TimeWindow{start=-4000, end=16000}
		// TimeWindow{start=0, end=20000}
		// TimeWindow{start=4000, end=24000}
		// TimeWindow{start=8000, end=28000}
		Collection<W> actualWindows = windowFunction.assignActualWindows(inputRow, timestamp);
		for (W window : actualWindows) {
			isElementDropped = false;
			triggerContext.window = window;
			// 判斷窗口是否立即觸發，例如earliy fire模式，默認這里是不觸發的，交給onEventTime()或onProcessingTime()來觸發
			boolean triggerResult = triggerContext.onElement(inputRow, timestamp);
			if (triggerResult) {
				emitWindowResult(window);
			}
			// 注冊清理時間，根據時間模式，分別對應到Event Time對應Timer或Processing Time對應Timer
			// Event Time對應Timer通過全局watermark來觸發，實現代碼在InternalTimerServiceImpl#advanceWatermark()
			// watermark是一個遞增的邏輯，后面代碼解析
			registerCleanupTimer(window);
		}

		if (isElementDropped) {
			// markEvent will increase numLateRecordsDropped
			lateRecordsDroppedRate.markEvent();
		}
	}
}

運行數據：

3、Timer觸發
I、InternalTimerServiceImpl#advanceWatermark()

WindowOperator#onEventTime()的調用前，可以先看其上層調用：InternalTimerServiceImpl#advanceWatermark()

當獲取的watermark為9999L時，把eventTimeTimerQueue隊列中所有小於這個值的timer poll出來，調用WindowOperator.onEnventTime(timer)

II、WindwOperator#onEventTime()

WindwOperator#onEventTime()方法比較清晰，主要是window的觸發和window的清理兩段邏輯：

public class WindowOperator{
	public void onEventTime(InternalTimer<K, W> timer) throws Exception {
		setCurrentKey(timer.getKey());

		triggerContext.window = timer.getNamespace();
		if (triggerContext.onEventTime(timer.getTimestamp())) {
			// fire
			emitWindowResult(triggerContext.window);
		}

		if (windowAssigner.isEventTime()) {
			windowFunction.cleanWindowIfNeeded(triggerContext.window, timer.getTimestamp());
		}
	}
}

III、emitWindowResult()提交結果

emitWindowResult()重點關注下其第一行代碼：

BaseRow aggResult = windowFunction.getWindowAggregationResult(window);
這個表示根據具體的TimeWindow{start=4000, end=24000}，去獲取聚合數據，如果是滑動窗口，需要將4000, 8000 ,12000，16000 , 20000, 24000這幾段affect窗口里面的聚合值合並起來，內部邏輯：

public class PanedWindowProcessFunction{
	public BaseRow getWindowAggregationResult(W window) throws Exception {
		Iterable<W> panes = windowAssigner.splitIntoPanes(window);
		BaseRow acc = windowAggregator.createAccumulators();
		// null namespace means use heap data views
		windowAggregator.setAccumulators(null, acc);
		for (W pane : panes) {
			BaseRow paneAcc = ctx.getWindowAccumulators(pane);
			if (paneAcc != null) {
				windowAggregator.merge(pane, paneAcc);
			}
		}
		return windowAggregator.getValue(window);
	}
}

Emit（Trigger）觸發器

• 配置方式指定Trigger：Flink1.9.0目前支持通過TableConifg配置earlyFireInterval、lateFireInterval毫秒數，來指定窗口結束之前、窗口結束之后的觸發策略（默認是watermark超過窗口結束后觸發一次），策略的解析在WindowEmitStrategy，在StreamExecGroupWindowAggregateRule就會創建和解析這個策略
• SQL方式指定Trigger：Flink1.9.0代碼中calcite部分已有SqlEmit相關的實現，后續可以支持SQL 語句（INSERT INTO）中配置EMIT觸發器

本文Emit和Trigger都是觸發器這一個概念，只是使用的方式不一樣

1、Emit策略
Emit 策略是指在Flink SQL 中，query的輸出策略（如能忍受的延遲）可能在不同的場景有不同的需求，而這部分需求，傳統的 ANSI SQL 並沒有對應的語法支持。比如用戶需求：1小時的時間窗口，窗口觸發之前希望每分鍾都能看到最新的結果，窗口觸發之后希望不丟失遲到一天內的數據。針對這類需求，抽象出了EMIT語法，並擴展到了SQL語法。

2、用途
EMIT語法的用途目前總結起來主要提供了：控制延遲、數據精確性，兩方面的功能。

• 控制延遲。針對大窗口，設置窗口觸發之前的EMIT輸出頻率，減少用戶看到結果的延遲(WITH| WITHOUT DELAY)。
• 數據精確性。不丟棄窗口觸發之后的遲到的數據，修正輸出結果(minIdleStateRetentionTime，在WindowEmitStrategy中生成allowLateness)。

在選擇EMIT策略時，還需要與處理開銷進行權衡。因為越低的輸出延遲、越高的數據精確性，都會帶來越高的計算開銷。

3、語法
EMIT 語法是用來定義輸出的策略，即是定義在輸出（INSERT INTO）上的動作。當未配置時，保持原有默認行為，即 window 只在 watermark 觸發時 EMIT 一個結果。

語法：
INSERT INTO tableName
query
EMIT strategy [, strategy]*

strategy ::= {WITH DELAY timeInterval | WITHOUT DELAY}
[BEFORE WATERMARK |AFTER WATERMARK]

timeInterval ::=‘string’ timeUnit

WITH DELAY：聲明能忍受的結果延遲，即按指定 interval 進行間隔輸出。
WITHOUT DELAY：聲明不忍受延遲，即每來一條數據就進行輸出。
BEFORE WATERMARK：窗口結束之前的策略配置，即watermark 觸發之前。
AFTER WATERMARK：窗口結束之后的策略配置，即watermark 觸發之后。
注：

• 其中 strategy可以定義多個，同時定義before和after的策略。 但不能同時定義兩個 before 或 兩個after 的策略。
• 若配置了AFTER WATERMARK 策略，需要顯式地在TableConfig中配置minIdleStateRetentionTime標識能忍受的最大遲到時間。
• minIdleStateRetentionTime在window中只影響窗口何時清除，不直接影響窗口何時觸發， 例如配置為3600000，最多容忍1小時的遲到數據，超過這個時間的數據會直接丟棄

4、示例
如果我們已經有一個TUMBLE（ctime, INTERVAL ‘1’ HOUR）的窗口，tumble_window 的輸出是需要等到一小時結束才能看到結果，我們希望能盡早能看到窗口的結果（即使是不完整的結果）。例如，我們希望每分鍾看到最新的窗口結果：
INSERT INTO result
SELECT * FROM tumble_window
EMIT WITH DELAY ‘1’ MINUTE BEFORE WATERMARK – 窗口結束之前，每隔1分鍾輸出一次更新結果

tumble_window 會忽略並丟棄窗口結束后到達的數據，而這部分數據對我們來說很重要，希望能統計進最終的結果里。而且我們知道我們的遲到數據不會太多，且遲到時間不會超過一天以上，並且希望收到遲到的數據立刻就更新結果：
INSERT INTO result
SELECT * FROM tumble_window
EMIT WITH DELAY ‘1’ MINUTE BEFORE WATERMARK,
WITHOUT DELAY AFTER WATERMARK --窗口結束之后，每條到達的數據都輸出

tEnv.getConfig.setIdleStateRetentionTime(Time.days(1), Time.days(2))//min、max，只有Time.days(1)這個參數直接對window生效

補充一下WITH DELAY '1’這種配置的周期觸發策略（即DELAY大於0），最后都是由ProcessingTime系統時間觸發：

class WindowEmitStrategy{
  private def createTriggerFromInterval(
      enableDelayEmit: Boolean,
      interval: Long): Option[Trigger[TimeWindow]] = {
    if (!enableDelayEmit) {
      None
    } else {
      if (interval > 0) {
       // 系統時間觸發，小於wm的所有timer都執行onProcessingTime()
        Some(ProcessingTimeTriggers.every(Duration.ofMillis(interval)))
      } else {
       // 為0則每條都觸發
        Some(ElementTriggers.every())
      }
    }
  }
}

5、Trigger類和結構關系
在源碼中，Window Trigger的實現子類有10個左右，需要結合上一個小節的EMIT SQL能更容易理清他們之間的關系，這里簡單介紹下：

• AfterEndOfWindow：這個就是沒配置任何EMIT策略時，默認的EvenTime、ProcTime
• Window觸發策略（即窗口結束后觸發一次）
• EveryElement：即delay=0，在processElement()時直接觸發，無論是在窗口結束之前或者窗口結束之后都觸發，且不再注冊timer
• AfterEndOfWindowNoLate：對應EMIT WITHOUT DELAY AFTER WATERMARK，窗口結束之前不輸出，窗口結束之后無延遲輸出
• AfterFirstElementPeriodic：對應WITH DELAY ‘1’ MINUTE BEFORE| AFTER WATERMARK，即按系統時間周期執行，由ProcessingTime系統時間周期觸發

聲明：本號所有文章除特殊注明，都為原創，公眾號讀者擁有優先閱讀權，未經作者本人允許不得轉載，否則追究侵權責任。

關注我的公眾號，后台回復【JAVAPDF】獲取200頁面試題！
5萬人關注的大數據成神之路，不來了解一下嗎？
5萬人關注的大數據成神之路，真的不來了解一下嗎？
5萬人關注的大數據成神之路，確定真的不來了解一下嗎？

歡迎您關注《大數據成神之路》