最系統(tǒng)掌握Flink CDC系列之實時抽取Oracle數(shù)據(jù)（排雷和調(diào)優(yōu)實踐）

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

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

試用環(huán)境：

Oracle：11.2.0.4.0（RAC 部署）

Flink：1.13.1

Hadoop：3.2.1

通過 Flink on Yarn 方式部署使用

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

根據(jù)官方文檔說明，在 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 觀察，發(fā)現(xiàn)無法正常連接 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 的服務(wù)名) 錯認為 SID。于是嘗試閱讀 Flink CDC 涉及到 Oracle Connector 的源碼，發(fā)現(xiàn)在 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 的連接方式并未做區(qū)分，而是直接在代碼中寫死了 SID 的連接方式 (即 port 和 dbname 中間使用 “ : ” 分隔開)。

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

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

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

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

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

二、無法找到 Oracle 表

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

[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，為什么數(shù)據(jù)庫名、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 的開發(fā)者將 “大小寫不敏感” 統(tǒng)一定義為了 “需要將表名轉(zhuǎn)換為小寫”。對于 Debezium 支持的 PostgreSQL、Mysql 等確實如此。然而對于 Oracle 數(shù)據(jù)庫，“大小寫不敏感” 卻意味著在內(nèi)部元信息存儲時，需要將表名轉(zhuǎn)換為大寫

因而 Debezium 在讀取到 “大小寫不敏感” 的配置后，按照上述代碼邏輯，只會因為嘗試去讀取小寫的表名而報錯。

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

如需使用 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 參數(shù)，Debezium 目前僅對 Oracle 11g 默認設(shè)置為 true，對其余 Oracle 版本均默認設(shè)置為 false。所以讀者如果使用的不是 Oracle 11g 版本，可無需修改該參數(shù)，但仍需顯式指定大寫的表名。

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

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

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

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

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

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