最系統掌握Flink CDC系列之即時擷取Oracle資料（排雷與調優實務）

這篇文章為大家帶來了對 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 更多的信息监控，以及由此带来的频繁的日志切换和日志转储操作，其代价也是惊人的。

根据笔者实际测试情况，如果 debezium.log.mining.strategy 为默认配置 redo_log_catalog，则不仅需要多执行第 ① 步操作 (该操作耗时约为半分钟到 1 分钟之间)，在第 ④ 步，根据 archived logs 的数据量，耗时也会在 1 分钟到 4 分钟之间浮动；在第 ⑤ 步，实际查询 V$LOGMNR_CONTENTS 视图也常常需要十几秒才能完成。

此外，由于 archive logs 在实际系统中增长速度较快，因此在实际使用中，常会配合进行定期删除或转储过期日志的操作。由于上述第 ④ 步的耗时较长，笔者观察到在第 ④ 步执行的过程中，在一定概率下会发生第 ② 步加入的a rchive logs 已过期而被删除转储的情况，于是在第 ⑤ 步查询的时候，会由于找不到第 ② 步加入的日志，而报下面的错误：

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

一般来说，Flink CDC 所需要监控的表，特别是对于业务系统有重大意义的表，一般不会进行 DDL 操作，仅需要捕捉 DML 操作即可，且对于数据库宕机等极特殊情况，也可使用在数据库恢复后进行全量数据更新的方式保障数据的一致性。因而，online_catalog 的方式足以满足我们的需要。

另外，无论使用 online_catalog，还是默认的 redo_log_catalog，都会存在第 ② 步找到的日志和第 ⑤ 步实际需要的日志不同步的问题，因此，加入 'debezium.log.mining.continuous.mine'='true' 参数，将实时搜集日志的工作交给 Oracle 自动完成，即可规避这一问题。

笔者按照这两个参数配置后，数据延迟一般可以从数分钟降至 5 秒钟左右。

四、调节参数继续降低数据延迟

上述流程图的第 ③ 步和第 ⑦ 步，提到了根据配置项来确定 LogMiner 监控时序范围，以及确定休眠时间。下面对该过程进行进一步分析，并对单个表的进一步调优给出一般性的方法论。

通过观察 io.debezium.connector.oracle.logminer.LogMinerHelper 类中的 getEndScn 方法，可了解到 debezium 对监控时序范围和休眠时间的调节原理。为便于读者理解，将该方法用流程图说明如下：

从上述的流程图中可以看出，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);
}

如果当前的监控时序范围达到了 log.mining.batch.size.max，那么 debezium 会在日志中给出如上提示。在实际应用中，观察 Flink CDC 产生的 log 是否包含该提示，便可得知 log.mining.batch.size.max 的值是否合理。

五、Debezium Oracle Connector 的隐藏参数

事實上從上文我們已經了解了兩個隱藏參數：debezium.database.tablename.case.insensitive (見第二節內容) 和debezium.log.mining.continuous.mine (見第三節） ，這兩個參數在Debezium 的官方文件中均未給出實際說明，但實際上可以使用。透過分析源碼，現給出Debezium Oracle Connector 的所有隱藏參數，以及其說明如下：

筆者認為除了上面我們已經用到的兩個參數以外，同樣值得重點關注的是log.mining.history.recorder.class 參數。由於該參數目前預設為io.debezium.connector.oracle.logminer.NeverHistoryRecorder，是一個空類，所以我們在分析Flink CDC 行為時，透過自訂實作io.debezium.connector.oracle.logminer.HistoryRecorder 介面的類別，可在不修改原始碼的情況下，實現對Flink CDC 行為的個人化監控。

推薦教學：《Oracle教學》

以上是最系統掌握Flink CDC系列之即時擷取Oracle資料（排雷與調優實務）的詳細內容。更多資訊請關注PHP中文網其他相關文章！