实时计算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暂不支持更新流作为输入。

实时计算引擎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），如果不正确处理，乱序可能会导致正确性问题。