Realtime Compute for Apache Flink：Flink SQL中Changelog事件亂序處理原理

Changelog介紹

在關聯式資料庫領域，MySQL使用binlog（二進位日誌）記錄資料庫中所有修改操作，包括INSERT、UPDATE和DELETE操作。類似地，Flink SQL中的Changelog主要記錄資料變化，以實現增量資料處理。

在MySQL中，binlog可以用於資料備份、恢複、同步和複製。通過讀取和解析binlog中的操作記錄，可以實現增量資料同步和複製。變更資料擷取（CDC）作為一種常用的資料同步技術，常被用於監控資料庫中的資料變化，並將其轉換為事件流進行即時處理。CDC工具可用於將關聯式資料庫中的資料變化即時傳輸到其他系統或資料倉儲，以支援即時分析和報告。當前常用的CDC工具包括Debezium和Maxwell。Flink通過FLINK-15331支援了CDC，可以即時地整合外部系統的CDC資料，並實現即時資料同步和分析。

Changelog事件產生和處理

Changelog介紹中提到的binlog和CDC是與Flink整合的外部Changelog資料來源，Flink SQL內部也會產生Changelog資料。為了區分事件是否為更新事件，我們將僅包含INSERT類型事件的Changelog稱為追加流或非更新流，而同時包含其他類型（例如UPDATE）事件的Changelog稱為更新流。Flink中的一些操作（如分組彙總和去重）可以產生更新事件，產生更新事件的操作通常會使用狀態，這類操作被稱為狀態運算元。需要注意的是，並非所有狀態運算元都支援處理更新流。例如，Over視窗彙總和Interval Join暫不支援更新流作為輸入。

Realtime Compute引擎VVR 6.0及以上版本的Query操作，對應的運行時運算元、是否支援處理更新流消費以及是否產生更新，詳情請參見Query操作運行時資訊說明。

Changelog的事件類型

FLINK-6047引入了回撤機制，使用INSERT和DELETE兩種事件類型（儘管資料來源僅支援INSERT事件），實現了流SQL運算元的累加式更新演算法。FLINK-16987以後，Changelog事件類型被重構為四種類型（如下），形成一個完整的Changelog事件類型體系，便於與CDC生態系統串連。

/**
 * A kind of row in a Changelog.
 */
@PublicEvolving
public enum RowKind {

	/**
	 * Insertion operation.
	 */
	INSERT,

	/**
	 * Previous content of an updated row.
	 */
	UPDATE_BEFORE,

	/**
	 * New content of an updated row.
	 */
	UPDATE_AFTER,

	/**
	 * Deletion operation.
	 */
	DELETE
}

Flink不使用包含UPDATE_BEFORE和UPDATE_AFTER的複合UPDATE事件類型的原因主要有兩個方面：

• 拆分的事件無論是何種事件類型（僅RowKind不同）都具有相同的事件結構，這使得序列化更簡單。如果使用複合UPDATE事件，那麼事件要麼是異構的，要麼是INSERT或DELETE事件對齊UPDATE事件（例如，INSERT事件僅含有UPDATE_AFTER，DELETE事件僅含有UPDATE_BEFORE）。

• 在分布式環境下，經常涉及資料shuffle（例如Join、彙總）。即使使用複合UPDATE事件，有時仍需將其拆分為單獨的DELETE和INSERT事件進行shuffle，例如下面的樣本。

樣本

下面是一個複合UPDATE事件必須拆分為DELETE和INSERT事件的情境樣本。本文後續也將圍繞此SQL作業樣本討論Changelog事件亂序問題並提供相應的解決方案。

-- CDC source tables:  s1 & s2
CREATE TEMPORARY TABLE s1 (
  id BIGINT, 
  level BIGINT,
  PRIMARY KEY(id) NOT ENFORCED
)WITH (...);

CREATE TEMPORARY TABLE s2 (
  id BIGINT, 
  attr VARCHAR, 
  PRIMARY KEY(id) NOT ENFORCED
)WITH (...);

-- sink table: t1
CREATE TEMPORARY TABLE t1 (
  id BIGINT, 
  level BIGINT, 
  attr VARCHAR,
  PRIMARY KEY(id) NOT ENFORCED
)WITH (...);

-- join s1 and s2 and insert the result into t1 
INSERT INTO t1
SELECT 
  s1.*, s2.attr
FROM s1 JOIN s2
ON s1.level = s2.id;

假設源表s1中id為1的記錄的Changelog在時間t0插入(id=1, level=10)，然後在時間t1將該行更新為(id=1, level=20)。這對應三個拆分事件：

源表s1的主鍵是id，但Join操作需要按level列進行shuffle（見子句ON）。

如果Join運算元的並發數為2，那麼以上三個事件可能會被發送到兩個任務中。即使使用複合UPDATE事件，它們也需要在shuffle階段拆分，來保證資料的平行處理。

亂序原因

假設樣本中表s2已有兩行資料進入Join運算元（+I（id=10，attr=a1)，+I（id=20，attr=b1）），Join運算子從表s1新接收到三個Changelog事件。在分布式環境中，實際的Join在兩個任務上平行處理，下遊運算元（樣本中為Sink任務）接收的事件序列可能情況如下所示。

情況1的事件序列與順序處理中的事件序列相同。情況2和情況3顯示了Changelog事件在Flink SQL中到達下遊運算元時的亂序情況。亂序情況可能會導致不正確的結果。在樣本中，結果表聲明的主鍵是id，外部儲存進行upsert更新時，在情況2和3中，如果沒有其他措施，將從外部儲存不正確地刪除id=1的行，而期望的結果是(id=1, level=20, attr='b1')。

使用SinkUpsertMaterializer解決

在樣本中，Join操作產生更新流，其中輸出包含INSERT事件（+I）和UPDATE事件（-U和+U），如果不正確處理，亂序可能會導致正確性問題。