Flink 实践 | Flink CDC 系列 - 实时抽取 Oracle 数据,排雷和调优实践
共 8216字,需浏览 17分钟
· 2021-12-17
摘要:本文作者为中国农业银行研发中心丁杨,在 Flink CDC 2.1 版本发布后第一时间下载使用,并成功实现了对 Oracle 的实时数据捕获以及性能调优,现将试用过程中的一些关键细节进行分享。主要内容包括:
无法连接数据库 无法找到 Oracle 表 数据延迟较大 调节参数继续降低数据延迟 Debezium Oracle Connector 的隐藏参数
Flink CDC
https://ververica.github.io/flink-cdc-connectors/release-2.1/content/connectors/oracle-cdc.html
Debezium
https://debezium.io/documentation/reference/1.5/connectors/oracle.html
一、无法连接数据库
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' );
[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
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 的 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);
该问题已提交至 Flink CDC Issue 701: https://github.com/ververica/flink-cdc-connectors/issues/701
二、无法找到 Oracle 表
[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
private TableId toLowerCaseIfNeeded(TableId tableId) {
return tableIdCaseInsensitive ? tableId.toLowercase() : tableId;
}
见 https://docs.oracle.com/cd/E11882_01/server.112/e41084/sql_elements008.htm “Nonquoted identifiers are not case sensitive. Oracle interprets them as uppercase”
如需使用 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' );
该问题已提交至 Flink CDC Issue 702: https://github.com/ververica/flink-cdc-connectors/issues/702
三、数据延迟较大
'debezium.log.mining.strategy'='online_catalog',
'debezium.log.mining.continuous.mine'='true'
ORA-00308: cannot open archive log '/path/to/archive/log/...'
ORA-27037: unable to obtain file status
四、调节参数继续降低数据延迟
从上述的流程图中可以看出,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);
}
五、Debezium Oracle Connector
的隐藏参数
所以我们在分析 Flink CDC 行为时,通过自定义实现 io.debezium.connector.oracle.logminer.HistoryRecorder 接口的类,可在不修改源码的情况下,实现对 Flink CDC 行为的个性化监控。
Flink-CDC 项目地址:
https://github.com/ververica/flink-cdc-connectors
相关文章
Flink CDC 系列 - 实现 MySQL 数据实时写入 Apache Doris
Flink CDC 系列 - 构建 MySQL 和 Postgres 上的 Streaming ETL Flink CDC 2.1 正式发布,稳定性大幅提升,新增 Oracle,MongoDB 支持
![](https://filescdn.proginn.com/7191f45d69c6b998136ae57ebcbd58e2/84226a534d1d9c264048f55653305586.webp)
戳我,查看更多技术干货~