本篇文章給大家?guī)?lái)了對(duì) Oracle 的實(shí)時(shí)數(shù)據(jù)捕獲以及性能調(diào)優(yōu)，將試用過程中的一些關(guān)鍵細(xì)節(jié)進(jìn)行分享，希望對(duì)大家有幫助。

Flink CDC 于 2021 年 11 月 15 日發(fā)布了最新版本 2.1，該版本通過引入內(nèi)置 Debezium 組件，增加了對(duì) Oracle 的支持。筆者第一時(shí)間下載了該版本進(jìn)行試用并成功實(shí)現(xiàn)了對(duì) Oracle 的實(shí)時(shí)數(shù)據(jù)捕獲以及性能調(diào)優(yōu)，現(xiàn)將試用過程中的一些關(guān)鍵細(xì)節(jié)進(jìn)行分享。

試用環(huán)境：

Oracle：11.2.0.4.0（RAC 部署）

Flink：1.13.1

Hadoop：3.2.1

通過 Flink on Yarn 方式部署使用

一、無(wú)法連接數(shù)據(jù)庫(kù)

根據(jù)官方文檔說(shuō)明，在 Flink SQL CLI 中輸入以下語(yǔ)句：

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 觀察，發(fā)現(xiàn)無(wú)法正常連接 Oracle，報(bào)錯(cuò)如下：

[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

從報(bào)錯(cuò)信息來(lái)看，可能是由于 Flink CDC 誤將連接信息中提供的 MY_SERVICE_NAME (Oracle 的服務(wù)名) 錯(cuò)認(rèn)為 SID。于是嘗試閱讀 Flink CDC 涉及到 Oracle Connector 的源碼，發(fā)現(xiàn)在 com.ververica.cdc.connectors.oracle.OracleValidator 中，對(duì)于 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); }

由上可以看出，在當(dāng)前版本的 Flink CDC 中，對(duì)于 SID 和 Service Name 的連接方式并未做區(qū)分，而是直接在代碼中寫死了 SID 的連接方式 (即 port 和 dbname 中間使用 “ : ” 分隔開)。

從 Oracle 8i 開始，Oracle 已經(jīng)引入了 Service Name 的概念以支持?jǐn)?shù)據(jù)庫(kù)的集群 (RAC) 部署，一個(gè) Service Name 可作為一個(gè)數(shù)據(jù)庫(kù)的邏輯概念，統(tǒng)一對(duì)該數(shù)據(jù)庫(kù)不同的 SID 實(shí)例的連接。據(jù)此，可以考慮以下兩種方式：

在 Flink CDC 的 create table 語(yǔ)句中，將 database-name 由 Service Name 替換成其中一個(gè) SID。該方式能解決連接問題，但無(wú)法適應(yīng)主流的 Oracle 集群部署的真實(shí)場(chǎng)景；

對(duì)該源碼進(jìn)行修改。具體可在新建工程中，重寫 com.ververica.cdc.connectors.oracle.OracleValidator 方法，修改為 Service Name 的連接方式 (即 port 和 dbname 中間使用 “ / ” 分隔開)，即：

"jdbc:oracle:thin:@" + hostname + ":" + port + "/" + dbname, userName, userpwd);

筆者采用的就是第二種方法，實(shí)現(xiàn)了正常連接數(shù)據(jù)庫(kù)的同時(shí)，保留對(duì) Oracle Service Name 特性的使用。

二、無(wú)法找到 Oracle 表

按照上述步驟，再次通過 select * from TEST 觀察，發(fā)現(xiàn)依然無(wú)法正常獲取數(shù)據(jù)，報(bào)錯(cuò)如下：

[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

觀察到錯(cuò)誤日志中提到的表是 MY_SERVICE_NAME.MY_SCHEMA.test，為什么數(shù)據(jù)庫(kù)名、Schema 名都是大寫，而表名是小寫？

注意到該錯(cuò)誤由 io.debezium 包報(bào)出，通過分析該包的源代碼 (通過 Flink CDC 的 pom.xml 文件可知，目前使用的是 debezium 1.5.4 版本) 可知，在 io.debezium.relational.Tables 中有如下代碼：

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

可見，Debezium 的開發(fā)者將 “大小寫不敏感” 統(tǒng)一定義為了 “需要將表名轉(zhuǎn)換為小寫”。對(duì)于 Debezium 支持的 PostgreSQL、Mysql 等確實(shí)如此。然而對(duì)于 Oracle 數(shù)據(jù)庫(kù)，“大小寫不敏感” 卻意味著在內(nèi)部元信息存儲(chǔ)時(shí)，需要將表名轉(zhuǎn)換為大寫

因而 Debezium 在讀取到 “大小寫不敏感” 的配置后，按照上述代碼邏輯，只會(huì)因?yàn)閲L試去讀取小寫的表名而報(bào)錯(cuò)。

由于 Debezium 直到目前最新的穩(wěn)定版本 1.7.1，以及最新的開發(fā)版本 1.8.0 都未修復(fù)該問題，我們可以通過以下兩種方法繞過該問題：

如需使用 Oracle “大小寫不敏感” 的特性，可直接修改源碼，將上述 toLowercase 修改為 toUppercase (這也是筆者選擇的方法)；

如果不愿意修改源碼，且無(wú)需使用 Oracle “大小寫不敏感” 的特性，可以在 create 語(yǔ)句中加上 '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' 中必須顯式指定大寫的表名。

需要注明的是，對(duì)于 database.tablename.case.insensitive 參數(shù)，Debezium 目前僅對(duì) Oracle 11g 默認(rèn)設(shè)置為 true，對(duì)其余 Oracle 版本均默認(rèn)設(shè)置為 false。所以讀者如果使用的不是 Oracle 11g 版本，可無(wú)需修改該參數(shù)，但仍需顯式指定大寫的表名。

三、數(shù)據(jù)延遲較大

數(shù)據(jù)延遲較大，有時(shí)需要 3-5 分鐘才能捕捉到數(shù)據(jù)變化。對(duì)于該問題，在 Flink CDC FAQ 中已給出了明確的解決方案：在 create 語(yǔ)句中加上如下兩個(gè)配置項(xiàng)：

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

那么為什么要這樣做呢？我們依然可以通過分析源碼和日志，結(jié)合 Oracle Logminer 的工作原理來(lái)加深對(duì)工具的理解。

對(duì) Logminer 的抽取工作，主要在 Debezium 的 io.debezium.connector.oracle.logminer.LogMinerStreamingChangeEventSource 中 execute 方法進(jìn)行。為節(jié)約篇幅，本文不列出實(shí)際的源碼，僅提煉出關(guān)鍵過程繪于下面的流程圖，有興趣的讀者可以對(duì)照該流程圖，結(jié)合實(shí)際源碼進(jìn)行分析：

采用 redo_log_catalog 的方式，可以監(jiān)控?cái)?shù)據(jù)表的 DDL 信息，且由于 archive logs 被永久保存到磁盤上，可以在數(shù)據(jù)庫(kù)宕機(jī)后依然正常獲取到宕機(jī)前的所有 DDL 和 DML 操作。但由于涉及到比 online catalog