站长资讯网本篇文章给大家带来了对 Oracle 的实时数据捕获以及性能调优，将试用过程中的一些关键细节进行分享，希望对大家有帮助。
Flink CDC 于 2021 年 11 月 15 日发布了最新版本 2.1，该版本通过引入内置 Debezium 组件，增加了对 Oracle 的支持。笔者第一时间下载了该版本进行试用并成功实现了对 Oracle 的实时数据捕获以及性能调优，现将试用过程中的一些关键细节进行分享。
试用环境：
Oracle：11.2.0.4.0（RAC 部署）
Flink：1.13.1
Hadoop：3.2.1
通过 Flink on Yarn 方式部署使用
一、无法连接数据库
根据官方文档说明，在 Flink SQL CLI 中输入以下语句：
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' );
之后尝试通过 select * from TEST 观察，发现无法正常连接 Oracle，报错如下：
[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
从报错信息来看，可能是由于 Flink CDC 误将连接信息中提供的 MY_SERVICE_NAME (Oracle 的服务名) 错认为 SID。于是尝试阅读 Flink CDC 涉及到 Oracle Connector 的源码，发现在 com.ververica.cdc.connectors.oracle.OracleValidator 中，对于 Oracle 连接的代码如下：
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); }
由上可以看出，在当前版本的 Flink CDC 中，对于 SID 和 Service Name 的连接方式并未做区分，而是直接在代码中写死了 SID 的连接方式 (即 port 和 dbname 中间使用 “ : ” 分隔开)。
从 Oracle 8i 开始，Oracle 已经引入了 Service Name 的概念以支持数据库的集群 (RAC) 部署，一个 Service Name 可作为一个数据库的逻辑概念，统一对该数据库不同的 SID 实例的连接。据此，可以考虑以下两种方式：
在 Flink CDC 的 create table 语句中，将 database-name 由 Service Name 替换成其中一个 SID。该方式能解决连接问题，但无法适应主流的 Oracle 集群部署的真实场景；
对该源码进行修改。具体可在新建工程中，重写 com.ververica.cdc.connectors.oracle.OracleValidator 方法，修改为 Service Name 的连接方式 (即 port 和 dbname 中间使用 “ / ” 分隔开)，即：
"jdbc:oracle:thin:@" + hostname + ":" + port + "/" + dbname, userName, userpwd);
笔者采用的就是第二种方法，实现了正常连接数据库的同时，保留对 Oracle Service Name 特性的使用。
二、无法找到 Oracle 表
按照上述步骤，再次通过 select * from TEST 观察，发现依然无法正常获取数据，报错如下：
[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
观察到错误日志中提到的表是 MY_SERVICE_NAME.MY_SCHEMA.test，为什么数据库名、Schema 名都是大写，而表名是小写？
注意到该错误由 io.debezium 包报出，通过分析该包的源代码 (通过 Flink CDC 的 pom.xml 文件可知，目前使用的是 debezium 1.5.4 版本) 可知，在 io.debezium.relational.Tables 中有如下代码：
private TableId toLowerCaseIfNeeded(TableId tableId) { return tableIdCaseInsensitive ? tableId.toLowercase() : tableId; }
可见，Debezium 的开发者将 “大小写不敏感” 统一定义为了 “需要将表名转换为小写”。对于 Debezium 支持的 PostgreSQL、Mysql 等确实如此。然而对于 Oracle 数据库，“大小写不敏感” 却意味着在内部元信息存储时，需要将表名转换为大写
因而 Debezium 在读取到 “大小写不敏感” 的配置后，按照上述代码逻辑，只会因为尝试去读取小写的表名而报错。
由于 Debezium 直到目前最新的稳定版本 1.7.1，以及最新的开发版本 1.8.0 都未修复该问题，我们可以通过以下两种方法绕过该问题：
如需使用 Oracle “大小写不敏感” 的特性，可直接修改源码，将上述 toLowercase 修改为 toUppercase (这也是笔者选择的方法)；
如果不愿意修改源码，且无需使用 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' );
该方法的弊端是丧失了 Oracle “大小写不敏感” 的特性，在 'table-name' 中必须显式指定大写的表名。
需要注明的是，对于 database.tablename.case.insensitive 参数，Debezium 目前仅对 Oracle 11g 默认设置为 true，对其余 Oracle 版本均默认设置为 false。所以读者如果使用的不是 Oracle 11g 版本，可无需修改该参数，但仍需显式指定大写的表名。
三、数据延迟较大
数据延迟较大，有时需要 3-5 分钟才能捕捉到数据变化。对于该问题，在 Flink CDC FAQ 中已给出了明确的解决方案：在 create 语句中加上如下两个配置项：
'debezium.log.mining.strategy'='online_catalog', 'debezium.log.mining.continuous.mine'='true'
那么为什么要这样做呢？我们依然可以通过分析源码和日志，结合 Oracle Logminer 的工作原理来加深对工具的理解。
对 Logminer 的抽取工作，主要在 Debezium 的 io.debezium.connector.oracle.logminer.LogMinerStreamingChangeEventSource 中 execute 方法进行。为节约篇幅，本文不列出实际的源码，仅提炼出关键过程绘于下面的流程图，有兴趣的读者可以对照该流程图，结合实际源码进行分析：
采用 redo_log_catalog 的方式，可以监控数据表的 DDL 信息，且由于 archive logs 被永久保存到磁盘上，可以在数据库宕机后依然正常获取到宕机前的所有 DDL 和 DML 操作。但由于涉及到比 online catalog