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暫不支持更新流作為輸入。

實時計算引擎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），如果不正確處理，亂序可能會導致正確性問題。

SinkUpsertMaterializer

根據唯一鍵與upsert鍵的內容，當Join輸出的是更新流且其upsert鍵與結果表主鍵不匹配時，需要一個中間步驟來消除亂序帶來的影響，以及基于結果表的主鍵產生新的主鍵對應的Changelog事件。Flink在Join算子和下游算子之間引入了SinkUpsertMaterializer算子（FLINK-20374）。

結合亂序原因中的Changelog事件，可以看到Changelog事件亂序遵循著一些規則。例如，對于一個特定的upsert鍵（或upsert鍵為空則表示所有列），事件ADD（+I、+U）總是在事件RETRACT（-D、-U）之前發生；即使涉及到數據shuffle，相同upsert鍵的一對匹配的Changelog事件也總是被相同的任務處理。這些規則也說明了為什么示例僅存在亂序原因中三個Changelog事件的組合。

SinkUpsertMaterializer就是基于上述規則實現的，其工作原理如下圖所示。SinkUpsertMaterializer在其狀態中維護了一個RowData列表。當SinkUpsertMaterializer被觸發，在處理輸入行時，它根據推斷的upsert鍵或整個行（如果upsert鍵為空）檢查狀態列表中是否存在相同的行。在ADD的情況下添加或更新狀態中的行，在RETRACT的情況下從狀態中刪除行。最后，它根據結果表的主鍵生成Changelog事件，更多詳細信息請參見SinkUpsertMaterializer源代碼。

通過SinkUpsertMaterializer，將示例中Join算子輸出的Changelog事件處理并轉換為結果表主鍵對應的Changelog事件，結果如下圖所示。根據SinkUpsertMaterializer的工作原理，在情況2中，處理-U(id=1，level=10，attr='a1')時，會將最后一行從狀態中移除，并向下游發送倒數第二行；在情況3中，當處理+U (id=1,level=20,attr='b1')時，SinkUpsertMaterializer會將其原樣發出，而當處理-U(id=1,level=10,attr='a1')時，將從狀態中刪除行而不發出任何事件。最終，通過SinkUpsertMaterializer算子情況2和3也會得到期望結果 (id=1,level=20,attr='b1')。