Oracle 基于logminer实时读 Connector 实现

[baisui1981]

https://nightlies.apache.org/flink/flink-cdc-docs-release-3.2/docs/connectors/flink-sources/oracle-cdc/

https://github.com/airbytehq/airbyte/blob/master/airbyte-integrations/connectors/source-oracle/src/main/resources/spec.json

[baisui1981]

Oracle CDC 存在的问题

CREATE TABLE DEBEZIUM."base" (
  "base_id"        NUMBER          NOT NULL,
  "start_time"     TIMESTAMP       NULL,
  "update_date"    DATE            NULL,
  "update_time"    TIMESTAMP       DEFAULT CURRENT_TIMESTAMP NOT NULL,
  "price"          NUMBER(5, 2)    NULL,
  "json_content"   CLOB            NULL,
  "col_blob"       BLOB            NULL,
  "col_text"       CLOB            NULL,
  CONSTRAINT PK_base PRIMARY KEY ("base_id")
)
LOB ("json_content", "col_blob", "col_text") STORE AS (TABLESPACE USERS);

通过以上 create table ddl 创建的 oracle base表，执行insert一条insert语句，注意json_content，col_blob，col_text 字段均有值，flink-cdc 使用log mining 的online_catalog 策略进行监听增量更新，在redo 日志文件中会生成insert 和 update类型的 event，为什么不能合并成一条insert 消息呢？

原因分析

在使用 Flink-CDC 监听 Oracle 数据库的增量更新时，如果执行了一条 INSERT 语句，并且该语句包含了对 LOB 字段（如 CLOB 或 BLOB）的赋值，可能会导致在重做日志（redo log）中生成多条事件（例如一条 INSERT 和一条或更多 UPDATE 类型的事件）。这种行为的原因主要与 Oracle 数据库处理 LOB 数据的方式有关。以下是详细的解释：

LOB 数据的特殊处理

1. 分步写入：
   当插入包含 LOB 字段的数据时，Oracle 可能会将 LOB 数据的写入分为多个步骤。首先，它会插入非 LOB 字段的数据，并为 LOB 字段预留一个空值或占位符。
   然后，在后续的步骤中，Oracle 再将实际的 LOB 数据写入到相应的存储位置。这可能导致额外的 UPDATE 操作被记录到重做日志中。
2. 补充日志记录配置：
   如果数据库没有正确配置补充日志记录（Supplemental Logging），尤其是针对 LOB 字段的补充日志，那么变更数据捕获工具（如 Debezium 或 Flink-CDC）可能无法准确地捕捉到所有变更，从而导致重复事件。
3. LogMiner 的解析方式：
   LogMiner 是 Oracle 提供的一种用于读取和解析重做日志文件的工具。当使用 LogMiner 的 online_catalog 策略进行监听时，它可能会根据日志中的不同操作类型分别生成事件，即使这些操作实际上是由同一个原始 SQL 语句引发的。

解决方案

为了确保 Flink-CDC 能够正确处理 LOB 字段并合并相关事件，你可以考虑以下措施：

1. 检查并调整补充日志记录设置：
   确保为包含 LOB 字段的表启用了适当的补充日志记录。对于 LOB 字段，建议启用全补充日志（Full Supplemental Logging）或者至少为这些字段启用补充日志。
   使用如下命令来确保所有列都被记录：

ALTER TABLE DEBEZIUM."base" ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

2. 优化 Flink-CDC 配置：
   确认 Flink-CDC 连接器是否正确配置了处理 LOB 字段的能力，包括开启 lob.fetch.enabled 和适当设置 lob.max.length 参数。
   查看 Flink-CDC 的文档，确认是否有任何特定于 LOB 处理的配置项或已知问题。

3. 评估 LogMiner 设置：
   如果你有权限调整 LogMiner 的配置，可以尝试修改其解析逻辑以更好地处理 LOB 数据的变化。不过，这通常需要深入了解 LogMiner 的工作机制，并且可能涉及复杂的调整。

4. 去重机制：
   在应用层面上实现一定的去重机制，确保即使收到了多条事件，也只处理一次实际的变更。例如，可以通过比较主键或其他唯一标识符来过滤掉重复的消息。

示例：合并事件的策略

如果你无法完全避免多条事件的产生，可以在 Flink 应用程序中实现逻辑来合并这些事件。比如，你可以设计一个状态管理器，它会跟踪每个记录的最新状态，并仅在检测到真正的变更时触发下游处理。

// 假设有一个状态管理器，它保存每个记录的最新版本
StateStore stateStore = ...;

// 当接收到新的变更事件时
for (ChangeRecord record : changeRecords) {
    // 根据主键查找现有状态
    Record existingRecord = stateStore.get(record.getKey());

    if (existingRecord == null || !existingRecord.equals(record)) {
        // 如果是新记录或确实发生了变更，则更新状态并发送给下游
        stateStore.put(record.getKey(), record);
        downstreamProcessor.process(record);
    }
}

通过上述方法，你可以尽量减少由于 LOB 数据处理而导致的冗余事件，并确保应用程序能够正确反映数据库中的真实变化。