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 数据处理而导致的冗余事件，并确保应用程序能够正确反映数据库中的真实变化。

[baisui1981]

已经完成

[baisui1981]

启动时报以错误

Caused by: java.io.IOException: io.debezium.DebeziumException: The db history topic or its content is fully or partially missing. Please check database history topic configuration and re-execute the snapshot.
	at org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceSplitReader.fetch(IncrementalSourceSplitReader.java:101)
	at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58)
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:165)
	... 6 more
Caused by: io.debezium.DebeziumException: The db history topic or its content is fully or partially missing. Please check database history topic configuration and re-execute the snapshot.
	at io.debezium.relational.HistorizedRelationalDatabaseSchema.recover(HistorizedRelationalDatabaseSchema.java:59)
	at org.apache.flink.cdc.connectors.oracle.source.reader.fetch.OracleSourceFetchTaskContext.validateAndLoadDatabaseHistory(OracleSourceFetchTaskContext.java:275)
	at org.apache.flink.cdc.connectors.oracle.source.reader.fetch.OracleSourceFetchTaskContext.configure(OracleSourceFetchTaskContext.java:118)
	at org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher.submitTask(IncrementalSourceStreamFetcher.java:84)
	at org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceSplitReader.submitStreamSplit(IncrementalSourceSplitReader.java:261)
	at org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceSplitReader.pollSplitRecords(IncrementalSourceSplitReader.java:153)
	at org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceSplitReader.fetch(IncrementalSourceSplitReader.java:98)
	... 8 more

[baisui1981]

基于docker 的可提供logminer 的oracle 的服务端搭建

docker 启动命令：

docker run -d --name=oracle -p 9090:8080 -p 1521:1521 -v /root/oracle/oradata:/u01/app/oracle quay.io/maksymbilenko/oracle-12c

确保本地目录有权限： 本地目录权限要打开：chmod -R 777 /root/oracle/oradata

按照flink-cdc 官网说明，打开redo logger 功能，链接：https://nightlies.apache.org/flink/flink-cdc-docs-release-3.2/docs/connectors/flink-sources/oracle-cdc/#for-non-cdb-database

Enable log archiving

(1.1). Connect to the database as DBA

ORACLE_SID=xe
export ORACLE_SID
sqlplus /nolog
CONNECT sys/oracle AS SYSDBA

(1.2). Enable log archiving

alter system set db_recovery_file_dest_size = 10G;

// 需要确保目录存在，且owner 为oracle
// mkdir -p /u01/app/oracle/oradata/recovery_area
// chown oracle.oracle /u01/app/oracle/oradata/recovery_area
alter system set db_recovery_file_dest = '/u01/app/oracle/oradata/recovery_area' scope=spfile;

shutdown immediate;
startup mount;
alter database archivelog;
alter database open;
Note:

Enable log archiving requires database restart, pay attention when try to do it
The archived logs will occupy a large amount of disk space, so consider clean the expired logs the periodically

(1.3). Check whether log archiving is enabled

-- Should now "Database log mode: Archive Mode"
archive log list;
Note:

Supplemental logging must be enabled for captured tables or the database in order for data changes to capture the before state of changed database rows. The following illustrates how to configure this on the table/database level.

-- Enable supplemental logging for a specific table:
ALTER TABLE inventory.customers ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
-- Enable supplemental logging for database
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
Create an Oracle user with permissions

(2.1). Create Tablespace

sqlplus sys/oracle@host:port/xe AS SYSDBA;
CREATE TABLESPACE logminer_tbs DATAFILE '/u01/app/oracle/oradata/xe/logminer_tbs.dbf' SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
exit;

(2.2). Create a user and grant permissions

sqlplus sys/oracle@localhost:1521/xe AS SYSDBA;
CREATE USER flinkuser IDENTIFIED BY flinkpw DEFAULT TABLESPACE LOGMINER_TBS QUOTA UNLIMITED ON LOGMINER_TBS;
GRANT CREATE SESSION TO flinkuser;
GRANT SET CONTAINER TO flinkuser;
GRANT SELECT ON V_$DATABASE to flinkuser;
GRANT FLASHBACK ANY TABLE TO flinkuser;
GRANT SELECT ANY TABLE TO flinkuser;
GRANT SELECT_CATALOG_ROLE TO flinkuser;
GRANT EXECUTE_CATALOG_ROLE TO flinkuser;
GRANT SELECT ANY TRANSACTION TO flinkuser;
GRANT LOGMINING TO flinkuser;
GRANT ANALYZE ANY TO flinkuser;

GRANT CREATE TABLE TO flinkuser;
-- need not to execute if set scan.incremental.snapshot.enabled=true(default)
GRANT LOCK ANY TABLE TO flinkuser;
GRANT ALTER ANY TABLE TO flinkuser;
GRANT CREATE SEQUENCE TO flinkuser;

GRANT EXECUTE ON DBMS_LOGMNR TO flinkuser;
GRANT EXECUTE ON DBMS_LOGMNR_D TO flinkuser;

GRANT SELECT ON V_$LOG TO flinkuser;
GRANT SELECT ON V_$LOG_HISTORY TO flinkuser;
GRANT SELECT ON V_$LOGMNR_LOGS TO flinkuser;
GRANT SELECT ON V_$LOGMNR_CONTENTS TO flinkuser;
GRANT SELECT ON V_$LOGMNR_PARAMETERS TO flinkuser;
GRANT SELECT ON V_$LOGFILE TO flinkuser;
GRANT SELECT ON V_$ARCHIVED_LOG TO flinkuser;
GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO flinkuser;
exit;