Flink CDC 提取记录变更时间作为事件时间和 Hudi 表的 precombine.field 以及1970-01-01 取值问题

博主历时三年精心创作的《大数据平台架构与原型实现：数据中台建设实战》一书现已由知名IT图书品牌电子工业出版社博文视点出版发行，点击《重磅推荐：建大数据平台太难了！给我发个工程原型吧！》了解图书详情，京东购书链接：https://item.jd.com/12677623.html，扫描左侧二维码进入京东手机购书页面。

CDC 数据中的记录变更时间标记着这条记录在数据库中执行对应操作（创建/更新/删除）的时间，可以说是天然的“事件时间”，特别是对于那些本身没有记录时间字段的表来说就更加合适了。Flink 官方文档 也建议在使用 CDC 的情况下，优先使用 CDC 中的这个时间字段，这个时间更加精准。

与此同时，在定义 Hudi 表时，precombine.field 也是一个非常重要的配置，显然 CDC 数据中的记录变更时间是最合适的，没有之一。

CDC 数据中的记录变更时间属于元数据范畴，以 Flink CDC 的 MySQL 数据库为例，它提供四种元数据的抽取：

Key	DataType	Description
table_name	STRING NOT NULL	Name of the table that contain the row.
database_name	STRING NOT NULL	Name of the database that contain the row.
op_ts	TIMESTAMP_LTZ(3) NOT NULL	It indicates the time that the change was made in the database. If the record is read from snapshot of the table instead of the binlog, the value is always 0.
row_kind	STRING NOT NULL	It indicates the row kind of the changelog,Note: The downstream SQL operator may fail to compare due to this new added column when processing the row retraction if the source operator chooses to output the ‘row_kind’ column for each record. It is recommended to use this metadata column only in simple synchronization jobs. ‘+I’ means INSERT message, ‘-D’ means DELETE message, ‘-U’ means UPDATE_BEFORE message and ‘+U’ means UPDATE_AFTER message.

其中的 op_ts 就是我们想要的，也就是：CDC 数据中的记录变更时间。我们可以在定义数据表时声明这个列，Flink CDC 可以将其提取出来作为普通字段供下游使用，就像下表中这样：

CREATE TABLE IF NOT EXISTS orders_mysql_cdc (`order_number` INT NOT NULL,`order_date` DATE NOT NULL,`purchaser` INT NOT NULL,`quantity` INT NOT NULL,`product_id` INT NOT NULL,`op_ts` TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,PRIMARY KEY (`order_number`) NOT ENFORCED
) WITH ('connector' = 'mysql-cdc',...
);

注意，在定义 Flink CDC 源表时，op_ts 的数据类型是 TIMESTAMP_LTZ(3)，不是 TIMESTAMP(3)，写入下游表时，可以是 TIMESTAMP(3)。

当我们初次使用这个 op_ts 字段时，你会发现，写入到的数据库的数据全部都是 1970-01-01 00:00:00.000，就像下面这样：

你可能会认为是哪里出错了，实际上，这是 Flink CDC 特别设计的，也是合理的，Flink CDC 官方文档的解释是：

If the record is read from snapshot of the table instead of the binlog, the value is always 0.

我们知道，Flink CDC ( 2.0+ ) 的一个显著特征是：它是全量 + 增量的一体化读取！全量就是经常说的历史数据，增量就是实时的数据，控制 Flink CDC 是从全部历史数据开始同步整个数据库还是从只当下的 binlog 中同步近期增量数据的配置项是：scan.startup.mode ( 官方文档 )，该配置项支持 5 种配置，而默认配置（initial）就是以当前分界点，数据中的现有数据使用全量方式读取（也叫快照读取），此后的数据从 binlog 中读取，这样就和上面描述的 op_ts 字段的取值吻合上了：

当 Flink CDC 使用全量方式读取表中的历史数据时，op_ts 字段全部取值为 0，即 1970-01-01 00:00:00.000，当 Flink CDC 使用增量方式读取 binlog 数据时，op_ts 字段的取值为数据发生变更的实际时间。

这种设计还是非常合理的，因为，Flink CDC 本身在使用快照方式读取时，就没有任何变更时间可以读取，这个时间只在 binlog 中才有，而这对下游也不会造成太大的影响，因为此时的数据都是 insert-only 的数据，同一主键也不会出现两条记录，至少对 Hudi 表是没有影响的。

此外，作为一个“额外收获”，你会发现：op_ts 这个字段本身恰好标记了一条记录是通过全量同步进来的，还是增量同步进来的！

---

补充：以下是 Flink CDC 官方文档对 scan.startup.mode 5 种同步模式的解释：

The config option scan.startup.mode specifies the startup mode for MySQL CDC consumer. The valid enumerations are:

• initial (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the latest binlog.
• earliest-offset: Skip snapshot phase and start reading binlog events from the earliest accessible binlog offset.
• latest-offset: Never to perform snapshot on the monitored database tables upon first startup, just read from the end of the binlog which means only have the changes since the connector was started.
• specific-offset: Skip snapshot phase and start reading binlog events from a specific offset. The offset could be specified with binlog filename and position, or a GTID set if GTID is enabled on server.
• timestamp: Skip snapshot phase and start reading binlog events from a specific timestamp.