Flink CDC 系列 - 实时抽取 Oracle 数据，排雷和调优实践

Flink CDC

https://ververica.github.io/flink-cdc-connectors/release-2.1/content/connectors/oracle-cdc.html

Debezium

https://debezium.io/documentation/reference/1.5/connectors/oracle.html

create table TEST (A string)
WITH ('connector'='oracle-cdc',
    'hostname'='10.230.179.125',
    'port'='1521',
    'username'='myname',
    'password'='***',
    'database-name'='MY_SERVICE_NAME',
    'schema-name'='MY_SCHEMA',
    'table-name'='TEST' );

复制

[ERROR] Could not execute SQL statement. Reason:
oracle.net.ns.NetException: Listener refused the connection with the following error:
ORA-12505, TNS:listener does not currently know of SID given in connect descriptor

复制

public static Connection openConnection(Properties properties) throws SQLException {
    DriverManager.registerDriver(new oracle.jdbc.OracleDriver());
    String hostname = properties.getProperty("database.hostname");
    String port = properties.getProperty("database.port");
    String dbname = properties.getProperty("database.dbname");
    String userName = properties.getProperty("database.user");
    String userpwd = properties.getProperty("database.password");
    return DriverManager.getConnection(
            "jdbc:oracle:thin:@" + hostname + ":" + port + ":" + dbname, userName, userpwd);
}

复制

1. 在 Flink CDC 的 create table 语句中，将 database-name 由 Service Name 替换成其中一个 SID。该方式能解决连接问题，但无法适应主流的 Oracle 集群部署的真实场景；
   
   
2. 对该源码进行修改。具体可在新建工程中，重写 com.ververica.cdc.connectors.oracle.OracleValidator 方法，修改为 Service Name 的连接方式 (即 port 和 dbname 中间使用 “ / ” 分隔开)，即：
   
   
   "jdbc:oracle:thin:@" + hostname + ":" + port + "/" + dbname, userName, userpwd);

该问题已提交至 Flink CDC Issue 701：

https://github.com/ververica/flink-cdc-connectors/issues/701

[ERROR] Could not execute SQL statement. Reason:
io.debezium.DebeziumException: Supplemental logging not configured for table MY_SERVICE_NAME.MY_SCHEMA.test.  Use command: ALTER TABLE MY_SCHEMA.test ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS

复制

private TableId toLowerCaseIfNeeded(TableId tableId) {
    return tableIdCaseInsensitive ? tableId.toLowercase() : tableId;
}

复制

见 https://docs.oracle.com/cd/E11882_01/server.112/e41084/sql_elements008.htm

“Nonquoted identifiers are not case sensitive. Oracle interprets them as uppercase”

1. 如需使用 Oracle “大小写不敏感” 的特性，可直接修改源码，将上述 toLowercase 修改为 toUppercase (这也是笔者选择的方法)；
   
   
2. 如果不愿意修改源码，且无需使用 Oracle “大小写不敏感” 的特性，可以在 create 语句中加上 'debezium.database.tablename.case.insensitive'='false'，如下示例：

create table TEST (A string)
WITH ('connector'='oracle-cdc',
    'hostname'='10.230.179.125',
    'port'='1521',
    'username'='myname',
    'password'='***',
    'database-name'='MY_SERVICE_NAME',
'schema-name'='MY_SCHEMA',
'table-name'='TEST',
'debezium.database.tablename.case.insensitive'='false' );
复制

该问题已提交至 Flink CDC Issue 702：

https://github.com/ververica/flink-cdc-connectors/issues/702

'debezium.log.mining.strategy'='online_catalog',
'debezium.log.mining.continuous.mine'='true'

复制

ORA-00308: cannot open archive log '/path/to/archive/log/...'
ORA-27037: unable to obtain file status

复制

从上述的流程图中可以看出，debezium 给出 log.mining.batch.size.* 和 log.mining.sleep.time.* 两组参数，就是为了让每一次 logMiner 运行的步长能够尽可能和数据库自身 SCN 增加的步长一致。由此可见：

• log.mining.batch.size.* 和 log.mining.sleep.time.* 参数的设定，和数据库整体的表现有关，和单个表的数据变化情况无关；
  
  
• log.mining.batch.size.default 不仅仅是监控时序范围的起始值，还是监控时序范围变化的阈值。所以如果要实现更灵活的监控时序范围调整，可考虑适当减小该参数；
  
  
• 由于每一次确定监控时序范围时，都会根据 topScn 和 currentScn 的大小来调整 sleepTime，所以为了实现休眠时间更灵活的调整，可考虑适当增大 log.mining.sleep.time.increment.ms；
  
  
• log.mining.batch.size.max 不能过小，否则会有监控时序范围永远无法追上数据库当前 SCN 的风险。为此，debezium 在 io.debezium.connector.oracle.OracleStreamingChangeEventSourceMetrics 中存在以下逻辑：

复制

if (currentBatchSize == batchSizeMax) {
    LOGGER.info("LogMiner is now using the maximum batch size {}. This could be indicative of large SCN gaps", currentBatchSize);
}
复制

所以我们在分析 Flink CDC 行为时，通过自定义实现 io.debezium.connector.oracle.logminer.HistoryRecorder 接口的类，可在不修改源码的情况下，实现对 Flink CDC 行为的个性化监控。

Flink-CDC 项目地址：

https://github.com/ververica/flink-cdc-connectors

相关文章

• Flink CDC 系列 - 实现 MySQL 数据实时写入 Apache Doris

  
  

• Flink CDC 系列 - 构建 MySQL 和 Postgres 上的 Streaming ETL
  
  
• Flink CDC 2.1 正式发布，稳定性大幅提升，新增 Oracle，MongoDB 支持

• Flink Forward Asia 2021 延期，线上相见
  
  
• 奖金翻倍！Flink Forward Asia Hackathon 最新参赛指南请查收

   戳我，查看更多技术干货～