Flink 实践 | 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 最新参赛指南请查收

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