前言
Flink CDC 于 2021 年 11 月 15 日發(fā)布了最新版本 2.1，該版本通過引入內(nèi)置 Debezium 組件，增加了對 Oracle 的支持。對該版本進(jìn)行試用并成功實(shí)現(xiàn)了對 Oracle 的實(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
問題
1、無法連接數(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ù)名) 錯認(rèn)為 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);
}
由上可以看出，在當(dāng)前版本的 Flink CDC 中，對于 SID 和 Service Name 的連接方式并未做區(qū)分，而是直接在代碼中寫死了 SID 的連接方式 (即 port 和 dbname 中間使用 “ : ” 分隔開)。
從 Oracle 8i 開始，Oracle 已經(jīng)引入了 Service Name 的概念以支持?jǐn)?shù)據(jù)庫的集群 (RAC) 部署，一個 Service Name 可作為一個數(shù)據(jù)庫的邏輯概念，統(tǒng)一對該數(shù)據(jù)庫不同的 SID 實(shí)例的連接。據(jù)此，可以考慮以下兩種方式：
- 在 Flink CDC 的 create table 語句中，將 database-name 由 Service Name 替換成其中一個 SID。該方式能解決連接問題，但無法適應(yīng)主流的 Oracle 集群部署的真實(shí)場景；
- 對該源碼進(jìn)行修改。具體可在新建工程中，重寫 com.ververica.cdc.connectors.oracle.OracleValidator 方法，修改為 Service Name 的連接方式 (即 port 和 dbname 中間使用 “ / ” 分隔開)，即：
“jdbc:oracle:thin:@” + hostname + “:” + port + “/” + dbname, userName, userpwd);
2、無法連接數(shù)據(jù)庫無法找到 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 等確實(shí)如此。然而對于 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 默認(rèn)設(shè)置為 true，對其余 Oracle 版本均默認(rèn)設(shè)置為 false。所以讀者如果使用的不是 Oracle 11g 版本，可無需修改該參數(shù)，但仍需顯式指定大寫的表名。
3、數(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 方法進(jìn)行。為節(jié)約篇幅，本文不列出實(shí)際的源碼，僅提煉出關(guān)鍵過程繪于下面的流程圖，有興趣的讀者可以對照該流程圖，結(jié)合實(shí)際源碼進(jìn)行分析：
采用 redo_log_catalog 的方式，可以監(jiān)控數(shù)據(jù)表的 DDL 信息，且由于 archive logs 被永久保存到磁盤上，可以在數(shù)據(jù)庫宕機(jī)后依然正常獲取到宕機(jī)前的所有 DDL 和 DML 操作。但由于涉及到比 online catalog 更多的信息監(jiān)控，以及由此帶來的頻繁的日志切換和日志轉(zhuǎn)儲操作，其代價也是驚人的。
一般來說，F(xiàn)link CDC 所需要監(jiān)控的表，特別是對于業(yè)務(wù)系統(tǒng)有重大意義的表，一般不會進(jìn)行 DDL 操作，僅需要捕捉 DML 操作即可，且對于數(shù)據(jù)庫宕機(jī)等極特殊情況，也可使用在數(shù)據(jù)庫恢復(fù)后進(jìn)行全量數(shù)據(jù)更新的方式保障數(shù)據(jù)的一致性。因而，online_catalog 的方式足以滿足我們的需要。
另外，無論使用 online_catalog，還是默認(rèn)的 redo_log_catalog，都會存在第 ② 步找到的日志和第 ⑤ 步實(shí)際需要的日志不同步的問題，因此，加入 ‘debezium.log.mining.continuous.mine’=‘true’ 參數(shù)，將實(shí)時搜集日志的工作交給 Oracle 自動完成，即可規(guī)避這一問題。
按照這兩個參數(shù)配置后，數(shù)據(jù)延遲一般可以從數(shù)分鐘降至 5 秒鐘左右。
4、調(diào)節(jié)參數(shù)繼續(xù)降低數(shù)據(jù)延遲
上述流程圖的第 ③ 步和第 ⑦ 步，提到了根據(jù)配置項來確定 LogMiner 監(jiān)控時序范圍，以及確定休眠時間。下面對該過程進(jìn)行進(jìn)一步分析，并對單個表的進(jìn)一步調(diào)優(yōu)給出一般性的方法論。
通過觀察 io.debezium.connector.oracle.logminer.LogMinerHelper 類中的 getEndScn 方法，可了解到 debezium 對監(jiān)控時序范圍和休眠時間的調(diào)節(jié)原理。為便于讀者理解，將該方法用流程圖說明如下：
從上述的流程圖中可以看出，debezium 給出 log.mining.batch.size.* 和 log.mining.sleep.time.* 兩組參數(shù)，就是為了讓每一次 logMiner 運(yùn)行的步長能夠盡可能和數(shù)據(jù)庫自身 SCN 增加的步長一致。由此可見：
- log.mining.batch.size.* 和 log.mining.sleep.time.* 參數(shù)的設(shè)定，和數(shù)據(jù)庫整體的表現(xiàn)有關(guān)，和單個表的數(shù)據(jù)變化情況無關(guān)；
- log.mining.batch.size.default 不僅僅是監(jiān)控時序范圍的起始值，還是監(jiān)控時序范圍變化的閾值。所以如果要實(shí)現(xiàn)更靈活的監(jiān)控時序范圍調(diào)整，可考慮適當(dāng)減小該參數(shù)；
- 由于每一次確定監(jiān)控時序范圍時，都會根據(jù) topScn 和 currentScn 的大小來調(diào)整 sleepTime，所以為了實(shí)現(xiàn)休眠時間更靈活的調(diào)整，可考慮適當(dāng)增大 log.mining.sleep.time.increment.ms；
- log.mining.batch.size.max 不能過小，否則會有監(jiān)控時序范圍永遠(yuǎn)無法追上數(shù)據(jù)庫當(dāng)前 SCN 的風(fēng)險。為此，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);
}
如果當(dāng)前的監(jiān)控時序范圍達(dá)到了 log.mining.batch.size.max，那么 debezium 會在日志中給出如上提示。在實(shí)際應(yīng)用中，觀察 Flink CDC 產(chǎn)生的 log 是否包含該提示，便可得知 log.mining.batch.size.max 的值是否合理。
5、Debezium Oracle Connector的隱藏參數(shù)
事實(shí)上從上文中我們已經(jīng)了解到了兩個隱藏參數(shù)：
debezium.database.tablename.case.insensitive (見第二節(jié)內(nèi)容)
debezium.log.mining.continuous.mine (見第三節(jié)內(nèi)容)
這兩個參數(shù)在 Debezium 的官方文檔中均未給出實(shí)際說明，但實(shí)際上可以使用。通過分析源碼，現(xiàn)給出 Debezium Oracle Connector 的所有隱藏參數(shù)，以及其說明如下：
除了上面我們已經(jīng)用到的兩個參數(shù)以外，同樣值得重點(diǎn)關(guān)注的是 log.mining.history.recorder.class 參數(shù)。
由于該參數(shù)目前默認(rèn)為 io.debezium.connector.oracle.logminer.NeverHistoryRecorder，是一個空類；
所以我們在分析 Flink CDC 行為時，通過自定義實(shí)現(xiàn) io.debezium.connector.oracle.logminer.HistoryRecorder 接口的類，可在不修改源碼的情況下，實(shí)現(xiàn)對 Flink CDC 行為的個性化監(jiān)控。文章來源：http://www.zghlxwxcb.cn/news/detail-793466.html
更多文章請掃碼關(guān)注公眾號，有問題的小伙伴也可以在公眾號上提出。
文章來源地址http://www.zghlxwxcb.cn/news/detail-793466.html
到了這里，關(guān)于Flink CDC 實(shí)時抽取 Oracle 數(shù)據(jù)-排錯&調(diào)優(yōu)的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
