文章目录一、 技术背景二、 关键技术1、 Oracle LogMiner2、 Chunjun 的 LogMiner 关键流程3、修复 Chunjun Oracle LogMiner 问题一、 技术背景在大数据实时同步场景中需要将Oracle 数据库的变更数据CDC采集并写入Apache Doris以支持数据分析、BI 报表、实时数据仓库等应用。本方案基于Flink Chunjun通过Oracle LogMiner 解析 Redo Log实现低延迟写入Doris。二、 关键技术1、 Oracle LogMinerLogMiner 是 Oracle 提供的 redo log 解析工具用于跟踪INSERT、UPDATE、DELETE操作。使用LogMiner需要现在Oracle中开启具体开启操作见Oracle配置LogMiner2、 Chunjun 的 LogMiner 关键流程Chunjun原 FlinkX是 Flink 生态的数据同步框架支持多种数据源连接器如 Oracle、MySQL、PostgreSQL、Doris。其中Chunjun Oracle LogMiner Source用于解析Oracle Redo Log 并转换为 Flink 数据流。如下整个流程架构Flink任务启动后通过Chunjun的oracle logMiner连接器建立 Oracle 连接启动 LogMiner 解析 Redo Log。实时监听V$LOGMNR_CONTENTS解析变更数据并转换为 Flink 事件流。具体地会将Oracle不同的操作日志解析为如下数据类型即重放数据操作Flink 任务处理数据完成转换、清洗等操作。Flink Sink 组件Chunjun Doris Sink将数据写入 Doris。操作类型before旧数据after新数据Flink 处理逻辑INSERT无{新数据}直接插入UPDATE{旧数据}{新数据}先删除旧数据再插入新数据DELETE{旧数据}无删除数据最后如下示例flink sqlCREATETABLEsource(IDint,NAME string)WITH(connectororaclelogminer-x,urljdbc:oracle:thin://xxx:1521/ORCL,usernamesystem,passwordxxx,catinsert,delete,update,tableTEST.TEST_USER,timestamp-format.standardSQL);CREATETABLEsink(k4int,k3 string)WITH(connectordoris-x,schemademo,passwordxxx,table-namemytable,urljdbc:mysql://xxx:9030,usernameroot,sink.parallelism1,lookup.error-limit100,lookup.cache-typeLRU,lookup.parallelism1,lookup.cache.ttl60000,lookup.cache.max-rows10000,writeModeUPSERT);insertintosinkselectIDask4,NAMEask3fromsource;3、修复 Chunjun Oracle LogMiner 问题在实际使用中Chunjun Oracle LogMiner 会遇到以下问题关于全量增量读数据的问题//LogMinerConfig没有全量同步的外部配置默认是增量读取数据privatebooleanenableFetchAlltrue;无法获取监听的表//LogMinerListener 中的LogMinerConfig没有set table的地方//即无法获取被监听的表改成直接获取logMinerConfig.getListenerTables();PavingData和Split 不能同时开启默认都开启将PavingData关闭