尧图网页设计 
SeaTunnel Oracle-CDC 连接器实战指南从 LogMiner 环境配置到 Exactly-Once 全增量同步【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel本文围绕 SeaTunnel 的 Oracle CDC 源连接器展开它基于 Oracle LogMiner 实现「快照 增量」两阶段数据捕获覆盖 CDC 任务在 Oracle 端的全部前置准备归档日志、补充日志、LogMiner 账户授权、完整的 Source 配置参数、数据类型映射以及自定义主键、Exactly-Once、按时间戳/SCN 启动、Schema 变更过滤等实战场景并结合 connector-cdc-oracle 模块源码解析其偏移量结构、启动模式校验与分块切分策略的底层实现。一、连接器定位与能力概览Oracle CDC 连接器插件名Oracle-CDC用于从 Oracle 数据库读取快照数据与增量数据通过 Debezium 嵌入式引擎驱动 LogMiner 挖掘 redo log 实现变更捕获。它同时支持 SeaTunnel Zeta 与 Flink 引擎从源码声明来看OracleIncrementalSource实现了SupportParallelism与SupportSchemaEvolution两个能力接口。能力是否支持批量batch否流式stream是exactly-once是列投影column projection否并行parallelism是用户自定义 split是一个重要的前提性说明文档原文 NoticeDebezium Oracle 连接器不依赖log.mining.continuous.mine连续挖掘选项——连接器自身负责检测日志切换并自动调整被挖掘的日志因此不能在debezium块中设置该属性。数据源驱动信息Datasource支持的版本DriverUrl 示例Oracle不同依赖版本对应不同驱动类oracle.jdbc.OracleDriverjdbc:oracle:thin:datasource01:1523:xe根据文档 FAQ 部分说明Oracle CDC 支持 Oracle Database11g、12c、19c、21c12c 及之后的多租户环境需使用 CDB root 连接配合 common userC##前缀。二、JDBC 驱动依赖安装Oracle CDC 的 JDBC 驱动不在连接器 jar 中需要按运行引擎分别放置Flink / Spark 引擎将 Oracle JDBC 驱动ojdbc8 放置到${SEATUNNEL_HOME}/plugins/目录如需支持 i18n 字符集将orai18n.jar一并复制到$SEATUNNEL_HOME/plugins/。SeaTunnel Zeta 引擎将 Oracle JDBC 驱动放置到${SEATUNNEL_HOME}/lib/目录同理orai18n.jar复制到$SEATUNNEL_HOME/lib/。从源码可以印证驱动加载机制OracleIncrementalSourceFactory 的restoreSource与 OracleSourceConfigFactory 的create方法中都有Class.forName(oracle.jdbc.OracleDriver)配置工厂内为oracle.jdbc.driver.OracleDriver全限定名的加载动作加载失败仅打印 warn——这也是驱动 jar 缺失时任务难以排查的典型根因。三、Oracle 数据库侧准备启用 LogMinerSeaTunnel 使用 Oracle 内置的 LogMiner 工具进行 CDC。数据库端需完成两类工作开启归档日志 补充日志以及创建具备 LogMiner 权限的采集账户。以下按部署形态分述继承自文档中的完整操作步骤。3.1 非 CDB容器数据库模式第 1 步在操作系统层面创建归档日志与用户表空间目录mkdir -p /opt/oracle/oradata/recovery_area mkdir -p /opt/oracle/oradata/ORCLCDB chown -R oracle /opt/oracle/***第 2 步以管理员登录并启用归档日志sqlplus /nolog; connect sys as sysdba; alter system set db_recovery_file_dest_size 10G; alter system set db_recovery_file_dest /opt/oracle/oradata/recovery_area scopespfile; shutdown immediate; startup mount; alter database archivelog; alter database open; ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS; archive log list;第 3 步创建logminer_user采集账户密码 oracle并授权读取表与日志CREATE TABLESPACE logminer_tbs DATAFILE /opt/oracle/oradata/ORCLCDB/logminer_tbs.dbf SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED; CREATE USER logminer_user IDENTIFIED BY oracle DEFAULT TABLESPACE logminer_tbs QUOTA UNLIMITED ON logminer_tbs; GRANT CREATE SESSION TO logminer_user; GRANT SELECT ON V_$DATABASE to logminer_user; GRANT SELECT ON V_$LOG TO logminer_user; GRANT SELECT ON V_$LOGFILE TO logminer_user; GRANT SELECT ON V_$LOGMNR_LOGS TO logminer_user; GRANT SELECT ON V_$LOGMNR_CONTENTS TO logminer_user; GRANT SELECT ON V_$ARCHIVED_LOG TO logminer_user; GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO logminer_user; GRANT EXECUTE ON DBMS_LOGMNR TO logminer_user; GRANT EXECUTE ON DBMS_LOGMNR_D TO logminer_user; GRANT SELECT ANY TRANSACTION TO logminer_user; GRANT SELECT ON V_$TRANSACTION TO logminer_user;注Oracle 11g 不支持GRANT LOGMINING语句仅需对采集表授权时使用GRANT SELECT ANY TABLE TO logminer_user; GRANT ANALYZE ANY TO logminer_user;3.2 CDB容器数据库 PDB可插拔数据库模式第 1 步创建目录mkdir -p /opt/oracle/oradata/recovery_area mkdir -p /opt/oracle/oradata/ORCLCDB mkdir -p /opt/oracle/oradata/ORCLCDB/ORCLPDB1 chown -R oracle /opt/oracle/***第 2 步以管理员启用日志sqlplus /nolog connect sys as sysdba; # Password: oracle alter system set db_recovery_file_dest_size 10G; alter system set db_recovery_file_dest /opt/oracle/oradata/recovery_area scopespfile; shutdown immediate startup mount alter database archivelog; alter database open; archive log list;第 3 步在 CDB 中为目标表启用补充日志ALTER TABLE TEST.* ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS; ALTER TABLE TEST.T2 ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;第 4 步分别创建采集账户使用的表空间。在 CDB 中sqlplus sys/top_secret//localhost:1521/ORCLCDB as sysdba CREATE TABLESPACE logminer_tbs DATAFILE /opt/oracle/oradata/ORCLCDB/logminer_tbs.dbf SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED; exit;在 PDB 中sqlplus sys/top_secret//localhost:1521/ORCLPDB1 as sysdba CREATE TABLESPACE logminer_tbs DATAFILE /opt/oracle/oradata/ORCLCDB/ORCLPDB1/logminer_tbs.dbf SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED; exit;第 5 步在 CDB 中创建 common userc##dbzuser并授予跨容器权限sqlplus sys/top_secret//localhost:1521/ORCLCDB as sysdba CREATE USER c##dbzuser IDENTIFIED BY dbz DEFAULT TABLESPACE logminer_tbs QUOTA UNLIMITED ON logminer_tbs CONTAINERALL; GRANT CREATE SESSION TO c##dbzuser CONTAINERALL; GRANT SET CONTAINER TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$DATABASE to c##dbzuser CONTAINERALL; GRANT FLASHBACK ANY TABLE TO c##dbzuser CONTAINERALL; GRANT SELECT ANY TABLE TO c##dbzuser CONTAINERALL; GRANT SELECT_CATALOG_ROLE TO c##dbzuser CONTAINERALL; GRANT EXECUTE_CATALOG_ROLE TO c##dbzuser CONTAINERALL; GRANT SELECT ANY TRANSACTION TO c##dbzuser CONTAINERALL; GRANT LOGMINING TO c##dbzuser CONTAINERALL; GRANT CREATE TABLE TO c##dbzuser CONTAINERALL; GRANT LOCK ANY TABLE TO c##dbzuser CONTAINERALL; GRANT CREATE SEQUENCE TO c##dbzuser CONTAINERALL; GRANT EXECUTE ON DBMS_LOGMNR TO c##dbzuser CONTAINERALL; GRANT EXECUTE ON DBMS_LOGMNR_D TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$LOG TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$LOG_HISTORY TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$LOGMNR_LOGS TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$LOGMNR_CONTENTS TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$LOGMNR_PARAMETERS TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$LOGFILE TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$ARCHIVED_LOG TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO c##dbzuser CONTAINERALL; exit;四、Source 配置参数详解以下为 Oracle-CDC 源插件的完整参数表含默认值与说明参数名类型必填默认值说明urlString是-JDBC 连接 URL例如jdbc:oracle:thin://oracle-host:1521/ORCLCDBusernameString是-连接 Oracle 数据库的用户名passwordString是-连接数据库服务器的密码database-namesList否-要监控的数据库名schema-namesList否-要监控的 schema 名table-namesList条件必填-要监控的表格式为database.schema.table例如ORCLCDB.DEBEZIUM.FULL_TYPES。与table-pattern二选一table-patternString条件必填-匹配表名的正则表达式。与table-names二选一table-names-configList否-按表配置例如[{table: ORCLCDB.DEBEZIUM.FULL_TYPES,primaryKeys: [ID],snapshotSplitColumn: ID}]。适用于表无主键、需要自定义主键或显式指定快照切分列的场景startup.modeEnum否INITIAL启动模式可选initial、latest、timestamp、specific。initial先同步历史快照再同步增量latest从最新位点开始、跳过初始快照timestamp从startup.timestamp换算的 SCN 开始specific从用户指定的 SCN 开始startup.timestampLong否-从指定时间戳Unix 纪元毫秒开始startup.mode timestamp时该时间戳会结合server-time-zone换算为 SCN。startup.mode为timestamp时必填startup.specific-offset.scnLong否-从指定 Oracle SCN 开始。startup.mode为specific时必填且该 SCN 必须仍可被所选 Oracle 日志挖掘后端访问stop.modeEnum否NEVER停止模式唯一合法值为never流式 Oracle CDC 源会持续运行直到任务被停止snapshot.split.sizeInteger否8096快照切分大小行数读表快照时表被切分为多个 splitsnapshot.fetch.sizeInteger否1024读表快照时每次 poll 的最大 fetch sizeserver-time-zoneString否UTC数据库服务器的会话时区未设置时使用ZoneId.systemDefault()同时用于将startup.timestamp换算为 SCN。数据库时区与 JVM 时区不一致时应显式设置connect.timeout.msDuration否30000连接数据库超时时间上限connect.max-retriesInteger否3建立数据库连接的最大重试次数connection.pool.sizeInteger否20JDBC 连接池大小incremental.parallelismInteger否1快照阶段结束后进入增量日志读取时使用的并行 reader 数chunk-key.even-distribution.factor.upper-boundDouble否100chunk 键分布因子上限。因子 (MAX(id) - MIN(id) 1) / 行数 小于等于该上界时按均匀分布优化切分否则视为不均匀分布若估算分片数超过sample-sharding.threshold则启用采样分片策略chunk-key.even-distribution.factor.lower-boundDouble否0.05chunk 键分布因子下限语义同上因子落在上下界之内按均匀分布切分之外触发采样策略sample-sharding.thresholdInteger否1000触发采样分片策略的估算分片数阈值估算行数 / chunk size 超过该值时启用采样分片用于更高效地处理超大表inverse-sampling.rateInteger否1000采样分片策略中的采样率倒数例如 1000 表示 1/1000 采样率控制最终分片数粒度split.allow-samplingBoolean否true设为 false 时无论分片数多少都回退到不等大 chunk 切分迭代查询方式enable_concurrent_readBoolean否true快照阶段是否启用基于 split 的并发读取设为 false 时源跳过 split 分析、将整表作为单个 split 读取适用于无索引表exactly_onceBoolean否false启用 Exactly-Once 语义use_select_countBoolean否false全量阶段用select count直接统计行数而非通过 analysis 表估算当 statistics 更新慢时直接 count 更快skip_analyzeBoolean否false全量阶段跳过表行数分析适用于已周期性调度 analysis 更新统计信息或数据不频繁变化的场景formatEnum否DEFAULT输出格式可选DEFAULT、COMPATIBLE_DEBEZIUM_JSONschema-changes.enabledBoolean否false默认禁用 Schema Evolution当前仅支持加列、删列、改列名、改列类型schema-changes.includeList否-仅让列出的 Schema 变更事件类型向下游传递需schema-changes.enabled true空表示全部放行schema-changes.excludeList否-列出的 Schema 变更事件类型不向下游传递在 include 之后生效冲突时exclude 优先debeziumConfig否-透传给 Debezium Embedded Engine 的属性用于捕获 Oracle 服务器数据变更common-options-否-源插件通用参数参见 Source Common Optionsdecimal_type_narrowingBoolean否true十进制类型窄化为 true 时在不损失精度的前提下将 decimal 窄化为 int 或 long 类型当前仅 Oracle 支持见下文详解4.1 启动模式的源码校验逻辑OracleIncrementalSourceFactory.optionRule 中定义了条件校验规则startup.specific-offset.scn仅在startup.mode specific时必需且必须大于 0startup.timestamp仅在timestamp模式时必需exactly_once仅在initial模式下可选。此外table-names与table-pattern被声明为互斥项exclusive。进一步地OracleIncrementalSource.getOracleStartupConfig 对specific模式做了显式处理SCN 会被封装成RedoLogOffset结构的scn、commit_scn0、lcr_positionnull三键位点而通用 CDC 基类中的 file/position 位点在 Oracle 下会被直接拒绝——因为 Oracle 的位点语义是 SCNSystem Change Number而非日志文件偏移。4.2 命名约束库名与表名必须大写从 OracleSourceConfigFactory.validateConfig 源码可见三条硬校验Oracle仅支持单个 databasedatabase.names列表只能有一个元素数据库名、表名中出现的字母必须全部为大写表名格式必须是${database}.${schema}.${table}或${schema}.${table}。配置示例中统一使用ORCLCDB.DEBEZIUM.FULL_TYPES这种三段式写法即是此原因的体现。4.3 decimal_type_narrowing 类型窄化decimal_type_narrowing true默认时不损失精度则窄化为整型false时保留 DECIMALOracletrue 时映射false 时映射NUMBER(1, 0)BooleanDecimal(1, 0)NUMBER(6, 0)INTDecimal(6, 0)NUMBER(10, 0)BIGINTDecimal(10, 0)4.4 快照行数统计的三种策略全量阶段需要先估算表行数来决定切分策略文档提供了三种可组合的手段默认通过 Oracle analysis 表统计信息估算行数use_select_count true直接执行select count(*)计数适用于统计信息过期、直接 count 更快的场景skip_analyze true跳过行数分析直接读取all_tables的NUM_ROWS适用于已周期性调度 analyze 更新统计信息、或数据不频繁变化的表。五、数据类型映射Oracle 类型SeaTunnel 类型INTEGERINTFLOATDECIMAL(38, 18)NUMBER(precision 9, scale 0)INTNUMBER(9 precision 18, scale 0)BIGINTNUMBER(18 precision, scale 0)DECIMAL(38, 0)NUMBER(precision 0, scale 0)DECIMAL(38, 18)NUMBER(scale ! 0)DECIMAL(38, 18)BINARY_DOUBLEDOUBLEBINARY_FLOAT / REALFLOATCHAR / NCHAR / NVARCHAR2 / VARCHAR2 / LONG / ROWID / NCLOB / CLOBSTRINGDATEDATETIMESTAMP / TIMESTAMP WITH LOCAL TIME ZONETIMESTAMPBLOB / RAW / LONG RAW / BFILEBYTES类型转换的实现位于 OracleTypeUtils它将 Debezium 的Column元信息类型名、长度、精度、小数位、默认值桥接为 SeaTunnel 的BasicTypeDefine再委托给OracleTypeConverter完成到 SeaTunnel 类型系统的转换其中对TIMESTAMP开头的类型会把length作为 scale 处理以保留时间精度信息。六、任务配置示例以下示例完整继承自官方文档可直接复制修改后使用。6.1 基础多表读取source { Oracle-CDC { plugin_output customers username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES, ORCLCDB.DEBEZIUM.FULL_TYPES2] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB source.reader.close.timeout 120000 connection.pool.size 1 debezium { database.oracle.jdbc.timezoneAsRegion false } } }6.2 用 select count 直接统计行数use_select_countsource { Oracle-CDC { plugin_output customers use_select_count true username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB source.reader.close.timeout 120000 } }6.3 读取 all_tables 的 NUM_ROWS 并跳过 analyzeskip_analyzesource { Oracle-CDC { plugin_output customers skip_analyze true username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB source.reader.close.timeout 120000 } }6.4 自定义主键table-names-configsource { Oracle-CDC { plugin_output customers url jdbc:oracle:thin://oracle-host:1521/ORCLCDB source.reader.close.timeout 120000 username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES] table-names-config [ { table ORCLCDB.DEBEZIUM.FULL_TYPES primaryKeys [ID] } ] } }6.5 启用 Exactly-Once CDCexactly_once true与默认startup.mode initial路径配合使用当下游 sink 也配置为 exactly-once 投递例如开启 XA 的 JDBC sink时启用env { parallelism 1 job.mode STREAMING checkpoint.interval 5000 } source { Oracle-CDC { plugin_output customers username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB exactly_once true connection.pool.size 1 debezium { database.oracle.jdbc.timezoneAsRegion false } } }6.6 从指定时间戳启动使用startup.mode timestamp从 Unix 毫秒时间戳换算出的 Oracle SCN 开始同步source { Oracle-CDC { plugin_output customers username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB startup.mode timestamp startup.timestamp 1700000000000 server-time-zone UTC debezium { database.oracle.jdbc.timezoneAsRegion false } } }6.7 配置 Debezium 心跳Heartbeat低流量场景下Oracle LogMiner 的 SCN 只有在 redo log 发生变化时才会前进。配置 Debezium 心跳可以让 SCN 持续移动使 checkpoint 位点被定期记录、复制延迟保持可观测。心跳表必须在任务启动前已在 Oracle 服务端存在source { Oracle-CDC { username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB debezium { database.oracle.jdbc.timezoneAsRegion false heartbeat.interval.ms 100 heartbeat.action.query INSERT INTO DEBEZIUM.heartbeat (ts) VALUES (SYSTIMESTAMP) } } }6.8 读取无主键表选择与源表保证程度匹配的路径仅追加Append-only负载下游永不产生 UPDATE/DELETE保持exactly_once false且不声明主键。源会退化为尽力而为的行标识没有可用键时连接器无法安全地应用 UPDATE/DELETE 事件。存在唯一非主键列通过table-names-config.primaryKeys声明该列并设置exactly_once true使快照阶段与 redo log 阶段使用同一把键保持行标识一致。source { Oracle-CDC { username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB table-names [ORCLCDB.DEBEZIUM.FULL_TYPES_NO_PRIMARY_KEY] table-names-config [ { table ORCLCDB.DEBEZIUM.FULL_TYPES_NO_PRIMARY_KEY primaryKeys [ID] } ] exactly_once true } }没有可用主键时连接器无法安全地应用 UPDATE/DELETE 事件该模式只应用于仅追加负载。6.9 Schema 变更事件过滤当schema-changes.enabled true时可用schema-changes.include/schema-changes.exclude进一步控制哪些变更事件向下游传播。使用 SeaTunnel 的规范事件名规范名操作add.column加列drop.column删列modify.column修改列类型/属性列名不变change.column重命名列可选地改类型update.columns上述四种列级变更的组别名优先级是确定性的① 若设置了schema-changes.include只有被包含的事件类型有资格下发② 随后应用schema-changes.exclude③ 同一类型同时出现在两个列表时exclude 胜出。source { Oracle-CDC { # ... schema-changes.enabled true schema-changes.include [add.column, drop.column] schema-changes.exclude [change.column] } }关于排除drop.column的数据处理注意对于被保留的NOT NULL列NULL写入会被 sink 拒绝——如果对某个源端已停止供数的 NOT NULL 列排除drop.column会在 sink 端失败。6.10 使用 Debezium 兼容格式发送到 Kafkaformat COMPATIBLE_DEBEZIUM_JSON必须与 Kafka sink 连接器配合使用细节参见 cdc-compatible-debezium-json 格式文档。七、源码级实现解析结合 connector-cdc-oracle 模块可以进一步理解文档中各项行为的底层支撑1. LogMiner 挖掘策略与只读模式。OracleSourceConfigFactory.create 会向 Debezium 属性集中写入若干 SeaTunnel 内部约定log.mining.read.only true只读挖掘模式对应 changelog 中的 ReadOnlyLogWriterFlushStrategy 支持tombstones.on.delete false禁用墓碑消息默认log.mining.strategy online_catalog而当schema-changes.enabled true时切换为redo_log_catalog——因为online_catalog无法正确捕获 DDL。源码中还有一处显式防御若用户在debezium块中把include.schema.changes设为 true 但同时指定了online_catalog策略会直接抛出 IllegalArgumentException。2. RedoLogOffset 位点结构。RedoLogOffset 用scn、commit_scn、lcr_position三个键描述 redo log 事件位点compareTo通过 Debezium 的Scn值做顺序比较NO_STOPPING_OFFSETLong.MIN_VALUE表示永不停止对应stop.mode never的流式语义。3. 快照切分ChunkSplitter。OracleChunkSplitter 实现了queryMinMax、sampleDataFromColumn调用OracleUtils.skipReadAndSortSampleData做跳读采样、queryNextChunkMax迭代式不等大分片与queryApproximateRowCnt等方法正是第四节中chunk-key.even-distribution.factor.*、inverse-sampling.rate、split.allow-sampling等参数的执行落点对ROWID类型切分键还使用了ROWID.compareBytes做字节级比较。enable_concurrent_read false时该切分分析整体被跳过、整表作为单 split 读取。4. Schema 变更解析。Oracle 的 DDL 变更由 OracleSchemaChangeResolver 委托给基于 ANTLR 的 CustomOracleAntlrDdlParser 解析 redo log 中的 DDL 语句产出AlterTableColumnEvent事件序列——这解释了为什么当前仅支持加列、删列、改名、改类型这四类列级变更与 OracleIncrementalSource.supports 中声明的ADD_COLUMN / DROP_COLUMN / RENAME_COLUMN / UPDATE_COLUMN一致。5. 任务分发。OracleDialect.createFetchTask 依据 split 类型分发到OracleSnapshotFetchTask快照阶段或OracleRedoLogFetchTask增量阶段内部复用 Debezium 的 LogMiner streaming event source 与只读日志写入冲刷策略 ReadOnlyLogWriterFlushStrategy。八、FAQQ1CDC 需要哪些 Oracle 权限LogMiner 用户需要以下权限GRANT CREATE SESSION TO logminer_user; GRANT SET CONTAINER TO logminer_user; GRANT SELECT ON V_$DATABASE TO logminer_user; GRANT FLASHBACK ANY TABLE TO logminer_user; GRANT SELECT ANY TABLE TO logminer_user; GRANT SELECT_CATALOG_ROLE TO logminer_user; GRANT EXECUTE_CATALOG_ROLE TO logminer_user; GRANT SELECT ANY TRANSACTION TO logminer_user; GRANT LOGMINING TO logminer_user; GRANT CREATE TABLE TO logminer_user; GRANT LOCK ANY TABLE TO logminer_user; GRANT CREATE SEQUENCE TO logminer_user; GRANT EXECUTE ON DBMS_LOGMNR TO logminer_user; GRANT EXECUTE ON DBMS_LOGMNR_D TO logminer_user; GRANT SELECT ON V_$LOG TO logminer_user; GRANT SELECT ON V_$LOG_HISTORY TO logminer_user; GRANT SELECT ON V_$LOGMNR_LOGS TO logminer_user; GRANT SELECT ON V_$LOGMNR_CONTENTS TO logminer_user; GRANT SELECT ON V_$LOGMNR_PARAMETERS TO logminer_user; GRANT SELECT ON V_$LOGFILE TO logminer_user; GRANT SELECT ON V_$ARCHIVED_LOG TO logminer_user; GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO logminer_user; GRANT SELECT ON V_$TRANSACTION TO logminer_user;同时在数据库级和表级启用补充日志ALTER DATABASE ADD SUPPLEMENTAL LOG DATA; ALTER TABLE schema_name.table_name ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;Q2支持多租户CDB/PDB数据库吗支持。将database-names设置为 CDB 名JDBC URL 指向 CDB root用户必须是 common userC##前缀且上述权限以CONTAINER ALL授予所有容器。Q3支持无主键表吗默认要求主键。若表中存在合适的唯一列可通过table-names-config的primaryKeys字段指定自定义主键列无可用主键时只能用于仅追加负载。Q4如何使用自定义快照查询通过debezium块内的 Debezium 属性snapshot.select.statement.overrides配置。该查询在 SeaTunnel 添加快照切分边界之前应用因此必须包含配置表 schema 及其切分键所需的全部列debezium { snapshot.select.statement.overrides DEBEZIUM.FULL_TYPES snapshot.select.statement.overrides.DEBEZIUM.FULL_TYPES SELECT * FROM DEBEZIUM.FULL_TYPES WHERE ACTIVE 1 }Q5如何提升 LogMiner 性能应将其首先当作数据库与 redo log 调优问题处理优先复用本文的 LogMiner 配置与补充日志小节仅为所需表启用日志在此基础上再考虑通过debezium透传属性调优且须先验证这些属性在你实际部署的 Oracle CDC 运行时版本中受支持。九、延伸阅读覆盖「全量 增量」同步完整生命周期、2PC sink 配置、Schema Evolution 与排障的生产级指南CDC Production Cookbook兼容 Debezium JSON 格式用于 Kafka sinkcdc-compatible-debezium-jsonOracle CDC 连接器变更历史connector-cdc-oracle changelog端到端测试connector-cdc-oracle-e2e【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 SeaTunnel Oracle-CDC 连接器实战指南：从 LogMiner 环境配置到 Exactly-Once 全增量同步
SeaTunnel Oracle-CDC 连接器实战指南从 LogMiner 环境配置到 Exactly-Once 全增量同步【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel本文围绕 SeaTunnel 的 Oracle CDC 源连接器展开它基于 Oracle LogMiner 实现「快照 增量」两阶段数据捕获覆盖 CDC 任务在 Oracle 端的全部前置准备归档日志、补充日志、LogMiner 账户授权、完整的 Source 配置参数、数据类型映射以及自定义主键、Exactly-Once、按时间戳/SCN 启动、Schema 变更过滤等实战场景并结合 connector-cdc-oracle 模块源码解析其偏移量结构、启动模式校验与分块切分策略的底层实现。一、连接器定位与能力概览Oracle CDC 连接器插件名Oracle-CDC用于从 Oracle 数据库读取快照数据与增量数据通过 Debezium 嵌入式引擎驱动 LogMiner 挖掘 redo log 实现变更捕获。它同时支持 SeaTunnel Zeta 与 Flink 引擎从源码声明来看OracleIncrementalSource实现了SupportParallelism与SupportSchemaEvolution两个能力接口。能力是否支持批量batch否流式stream是exactly-once是列投影column projection否并行parallelism是用户自定义 split是一个重要的前提性说明文档原文 NoticeDebezium Oracle 连接器不依赖log.mining.continuous.mine连续挖掘选项——连接器自身负责检测日志切换并自动调整被挖掘的日志因此不能在debezium块中设置该属性。数据源驱动信息Datasource支持的版本DriverUrl 示例Oracle不同依赖版本对应不同驱动类oracle.jdbc.OracleDriverjdbc:oracle:thin:datasource01:1523:xe根据文档 FAQ 部分说明Oracle CDC 支持 Oracle Database11g、12c、19c、21c12c 及之后的多租户环境需使用 CDB root 连接配合 common userC##前缀。二、JDBC 驱动依赖安装Oracle CDC 的 JDBC 驱动不在连接器 jar 中需要按运行引擎分别放置Flink / Spark 引擎将 Oracle JDBC 驱动ojdbc8 放置到${SEATUNNEL_HOME}/plugins/目录如需支持 i18n 字符集将orai18n.jar一并复制到$SEATUNNEL_HOME/plugins/。SeaTunnel Zeta 引擎将 Oracle JDBC 驱动放置到${SEATUNNEL_HOME}/lib/目录同理orai18n.jar复制到$SEATUNNEL_HOME/lib/。从源码可以印证驱动加载机制OracleIncrementalSourceFactory 的restoreSource与 OracleSourceConfigFactory 的create方法中都有Class.forName(oracle.jdbc.OracleDriver)配置工厂内为oracle.jdbc.driver.OracleDriver全限定名的加载动作加载失败仅打印 warn——这也是驱动 jar 缺失时任务难以排查的典型根因。三、Oracle 数据库侧准备启用 LogMinerSeaTunnel 使用 Oracle 内置的 LogMiner 工具进行 CDC。数据库端需完成两类工作开启归档日志 补充日志以及创建具备 LogMiner 权限的采集账户。以下按部署形态分述继承自文档中的完整操作步骤。3.1 非 CDB容器数据库模式第 1 步在操作系统层面创建归档日志与用户表空间目录mkdir -p /opt/oracle/oradata/recovery_area mkdir -p /opt/oracle/oradata/ORCLCDB chown -R oracle /opt/oracle/***第 2 步以管理员登录并启用归档日志sqlplus /nolog; connect sys as sysdba; alter system set db_recovery_file_dest_size 10G; alter system set db_recovery_file_dest /opt/oracle/oradata/recovery_area scopespfile; shutdown immediate; startup mount; alter database archivelog; alter database open; ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS; archive log list;第 3 步创建logminer_user采集账户密码 oracle并授权读取表与日志CREATE TABLESPACE logminer_tbs DATAFILE /opt/oracle/oradata/ORCLCDB/logminer_tbs.dbf SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED; CREATE USER logminer_user IDENTIFIED BY oracle DEFAULT TABLESPACE logminer_tbs QUOTA UNLIMITED ON logminer_tbs; GRANT CREATE SESSION TO logminer_user; GRANT SELECT ON V_$DATABASE to logminer_user; GRANT SELECT ON V_$LOG TO logminer_user; GRANT SELECT ON V_$LOGFILE TO logminer_user; GRANT SELECT ON V_$LOGMNR_LOGS TO logminer_user; GRANT SELECT ON V_$LOGMNR_CONTENTS TO logminer_user; GRANT SELECT ON V_$ARCHIVED_LOG TO logminer_user; GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO logminer_user; GRANT EXECUTE ON DBMS_LOGMNR TO logminer_user; GRANT EXECUTE ON DBMS_LOGMNR_D TO logminer_user; GRANT SELECT ANY TRANSACTION TO logminer_user; GRANT SELECT ON V_$TRANSACTION TO logminer_user;注Oracle 11g 不支持GRANT LOGMINING语句仅需对采集表授权时使用GRANT SELECT ANY TABLE TO logminer_user; GRANT ANALYZE ANY TO logminer_user;3.2 CDB容器数据库 PDB可插拔数据库模式第 1 步创建目录mkdir -p /opt/oracle/oradata/recovery_area mkdir -p /opt/oracle/oradata/ORCLCDB mkdir -p /opt/oracle/oradata/ORCLCDB/ORCLPDB1 chown -R oracle /opt/oracle/***第 2 步以管理员启用日志sqlplus /nolog connect sys as sysdba; # Password: oracle alter system set db_recovery_file_dest_size 10G; alter system set db_recovery_file_dest /opt/oracle/oradata/recovery_area scopespfile; shutdown immediate startup mount alter database archivelog; alter database open; archive log list;第 3 步在 CDB 中为目标表启用补充日志ALTER TABLE TEST.* ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS; ALTER TABLE TEST.T2 ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;第 4 步分别创建采集账户使用的表空间。在 CDB 中sqlplus sys/top_secret//localhost:1521/ORCLCDB as sysdba CREATE TABLESPACE logminer_tbs DATAFILE /opt/oracle/oradata/ORCLCDB/logminer_tbs.dbf SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED; exit;在 PDB 中sqlplus sys/top_secret//localhost:1521/ORCLPDB1 as sysdba CREATE TABLESPACE logminer_tbs DATAFILE /opt/oracle/oradata/ORCLCDB/ORCLPDB1/logminer_tbs.dbf SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED; exit;第 5 步在 CDB 中创建 common userc##dbzuser并授予跨容器权限sqlplus sys/top_secret//localhost:1521/ORCLCDB as sysdba CREATE USER c##dbzuser IDENTIFIED BY dbz DEFAULT TABLESPACE logminer_tbs QUOTA UNLIMITED ON logminer_tbs CONTAINERALL; GRANT CREATE SESSION TO c##dbzuser CONTAINERALL; GRANT SET CONTAINER TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$DATABASE to c##dbzuser CONTAINERALL; GRANT FLASHBACK ANY TABLE TO c##dbzuser CONTAINERALL; GRANT SELECT ANY TABLE TO c##dbzuser CONTAINERALL; GRANT SELECT_CATALOG_ROLE TO c##dbzuser CONTAINERALL; GRANT EXECUTE_CATALOG_ROLE TO c##dbzuser CONTAINERALL; GRANT SELECT ANY TRANSACTION TO c##dbzuser CONTAINERALL; GRANT LOGMINING TO c##dbzuser CONTAINERALL; GRANT CREATE TABLE TO c##dbzuser CONTAINERALL; GRANT LOCK ANY TABLE TO c##dbzuser CONTAINERALL; GRANT CREATE SEQUENCE TO c##dbzuser CONTAINERALL; GRANT EXECUTE ON DBMS_LOGMNR TO c##dbzuser CONTAINERALL; GRANT EXECUTE ON DBMS_LOGMNR_D TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$LOG TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$LOG_HISTORY TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$LOGMNR_LOGS TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$LOGMNR_CONTENTS TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$LOGMNR_PARAMETERS TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$LOGFILE TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$ARCHIVED_LOG TO c##dbzuser CONTAINERALL; GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO c##dbzuser CONTAINERALL; exit;四、Source 配置参数详解以下为 Oracle-CDC 源插件的完整参数表含默认值与说明参数名类型必填默认值说明urlString是-JDBC 连接 URL例如jdbc:oracle:thin://oracle-host:1521/ORCLCDBusernameString是-连接 Oracle 数据库的用户名passwordString是-连接数据库服务器的密码database-namesList否-要监控的数据库名schema-namesList否-要监控的 schema 名table-namesList条件必填-要监控的表格式为database.schema.table例如ORCLCDB.DEBEZIUM.FULL_TYPES。与table-pattern二选一table-patternString条件必填-匹配表名的正则表达式。与table-names二选一table-names-configList否-按表配置例如[{table: ORCLCDB.DEBEZIUM.FULL_TYPES,primaryKeys: [ID],snapshotSplitColumn: ID}]。适用于表无主键、需要自定义主键或显式指定快照切分列的场景startup.modeEnum否INITIAL启动模式可选initial、latest、timestamp、specific。initial先同步历史快照再同步增量latest从最新位点开始、跳过初始快照timestamp从startup.timestamp换算的 SCN 开始specific从用户指定的 SCN 开始startup.timestampLong否-从指定时间戳Unix 纪元毫秒开始startup.mode timestamp时该时间戳会结合server-time-zone换算为 SCN。startup.mode为timestamp时必填startup.specific-offset.scnLong否-从指定 Oracle SCN 开始。startup.mode为specific时必填且该 SCN 必须仍可被所选 Oracle 日志挖掘后端访问stop.modeEnum否NEVER停止模式唯一合法值为never流式 Oracle CDC 源会持续运行直到任务被停止snapshot.split.sizeInteger否8096快照切分大小行数读表快照时表被切分为多个 splitsnapshot.fetch.sizeInteger否1024读表快照时每次 poll 的最大 fetch sizeserver-time-zoneString否UTC数据库服务器的会话时区未设置时使用ZoneId.systemDefault()同时用于将startup.timestamp换算为 SCN。数据库时区与 JVM 时区不一致时应显式设置connect.timeout.msDuration否30000连接数据库超时时间上限connect.max-retriesInteger否3建立数据库连接的最大重试次数connection.pool.sizeInteger否20JDBC 连接池大小incremental.parallelismInteger否1快照阶段结束后进入增量日志读取时使用的并行 reader 数chunk-key.even-distribution.factor.upper-boundDouble否100chunk 键分布因子上限。因子 (MAX(id) - MIN(id) 1) / 行数 小于等于该上界时按均匀分布优化切分否则视为不均匀分布若估算分片数超过sample-sharding.threshold则启用采样分片策略chunk-key.even-distribution.factor.lower-boundDouble否0.05chunk 键分布因子下限语义同上因子落在上下界之内按均匀分布切分之外触发采样策略sample-sharding.thresholdInteger否1000触发采样分片策略的估算分片数阈值估算行数 / chunk size 超过该值时启用采样分片用于更高效地处理超大表inverse-sampling.rateInteger否1000采样分片策略中的采样率倒数例如 1000 表示 1/1000 采样率控制最终分片数粒度split.allow-samplingBoolean否true设为 false 时无论分片数多少都回退到不等大 chunk 切分迭代查询方式enable_concurrent_readBoolean否true快照阶段是否启用基于 split 的并发读取设为 false 时源跳过 split 分析、将整表作为单个 split 读取适用于无索引表exactly_onceBoolean否false启用 Exactly-Once 语义use_select_countBoolean否false全量阶段用select count直接统计行数而非通过 analysis 表估算当 statistics 更新慢时直接 count 更快skip_analyzeBoolean否false全量阶段跳过表行数分析适用于已周期性调度 analysis 更新统计信息或数据不频繁变化的场景formatEnum否DEFAULT输出格式可选DEFAULT、COMPATIBLE_DEBEZIUM_JSONschema-changes.enabledBoolean否false默认禁用 Schema Evolution当前仅支持加列、删列、改列名、改列类型schema-changes.includeList否-仅让列出的 Schema 变更事件类型向下游传递需schema-changes.enabled true空表示全部放行schema-changes.excludeList否-列出的 Schema 变更事件类型不向下游传递在 include 之后生效冲突时exclude 优先debeziumConfig否-透传给 Debezium Embedded Engine 的属性用于捕获 Oracle 服务器数据变更common-options-否-源插件通用参数参见 Source Common Optionsdecimal_type_narrowingBoolean否true十进制类型窄化为 true 时在不损失精度的前提下将 decimal 窄化为 int 或 long 类型当前仅 Oracle 支持见下文详解4.1 启动模式的源码校验逻辑OracleIncrementalSourceFactory.optionRule 中定义了条件校验规则startup.specific-offset.scn仅在startup.mode specific时必需且必须大于 0startup.timestamp仅在timestamp模式时必需exactly_once仅在initial模式下可选。此外table-names与table-pattern被声明为互斥项exclusive。进一步地OracleIncrementalSource.getOracleStartupConfig 对specific模式做了显式处理SCN 会被封装成RedoLogOffset结构的scn、commit_scn0、lcr_positionnull三键位点而通用 CDC 基类中的 file/position 位点在 Oracle 下会被直接拒绝——因为 Oracle 的位点语义是 SCNSystem Change Number而非日志文件偏移。4.2 命名约束库名与表名必须大写从 OracleSourceConfigFactory.validateConfig 源码可见三条硬校验Oracle仅支持单个 databasedatabase.names列表只能有一个元素数据库名、表名中出现的字母必须全部为大写表名格式必须是${database}.${schema}.${table}或${schema}.${table}。配置示例中统一使用ORCLCDB.DEBEZIUM.FULL_TYPES这种三段式写法即是此原因的体现。4.3 decimal_type_narrowing 类型窄化decimal_type_narrowing true默认时不损失精度则窄化为整型false时保留 DECIMALOracletrue 时映射false 时映射NUMBER(1, 0)BooleanDecimal(1, 0)NUMBER(6, 0)INTDecimal(6, 0)NUMBER(10, 0)BIGINTDecimal(10, 0)4.4 快照行数统计的三种策略全量阶段需要先估算表行数来决定切分策略文档提供了三种可组合的手段默认通过 Oracle analysis 表统计信息估算行数use_select_count true直接执行select count(*)计数适用于统计信息过期、直接 count 更快的场景skip_analyze true跳过行数分析直接读取all_tables的NUM_ROWS适用于已周期性调度 analyze 更新统计信息、或数据不频繁变化的表。五、数据类型映射Oracle 类型SeaTunnel 类型INTEGERINTFLOATDECIMAL(38, 18)NUMBER(precision 9, scale 0)INTNUMBER(9 precision 18, scale 0)BIGINTNUMBER(18 precision, scale 0)DECIMAL(38, 0)NUMBER(precision 0, scale 0)DECIMAL(38, 18)NUMBER(scale ! 0)DECIMAL(38, 18)BINARY_DOUBLEDOUBLEBINARY_FLOAT / REALFLOATCHAR / NCHAR / NVARCHAR2 / VARCHAR2 / LONG / ROWID / NCLOB / CLOBSTRINGDATEDATETIMESTAMP / TIMESTAMP WITH LOCAL TIME ZONETIMESTAMPBLOB / RAW / LONG RAW / BFILEBYTES类型转换的实现位于 OracleTypeUtils它将 Debezium 的Column元信息类型名、长度、精度、小数位、默认值桥接为 SeaTunnel 的BasicTypeDefine再委托给OracleTypeConverter完成到 SeaTunnel 类型系统的转换其中对TIMESTAMP开头的类型会把length作为 scale 处理以保留时间精度信息。六、任务配置示例以下示例完整继承自官方文档可直接复制修改后使用。6.1 基础多表读取source { Oracle-CDC { plugin_output customers username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES, ORCLCDB.DEBEZIUM.FULL_TYPES2] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB source.reader.close.timeout 120000 connection.pool.size 1 debezium { database.oracle.jdbc.timezoneAsRegion false } } }6.2 用 select count 直接统计行数use_select_countsource { Oracle-CDC { plugin_output customers use_select_count true username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB source.reader.close.timeout 120000 } }6.3 读取 all_tables 的 NUM_ROWS 并跳过 analyzeskip_analyzesource { Oracle-CDC { plugin_output customers skip_analyze true username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB source.reader.close.timeout 120000 } }6.4 自定义主键table-names-configsource { Oracle-CDC { plugin_output customers url jdbc:oracle:thin://oracle-host:1521/ORCLCDB source.reader.close.timeout 120000 username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES] table-names-config [ { table ORCLCDB.DEBEZIUM.FULL_TYPES primaryKeys [ID] } ] } }6.5 启用 Exactly-Once CDCexactly_once true与默认startup.mode initial路径配合使用当下游 sink 也配置为 exactly-once 投递例如开启 XA 的 JDBC sink时启用env { parallelism 1 job.mode STREAMING checkpoint.interval 5000 } source { Oracle-CDC { plugin_output customers username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB exactly_once true connection.pool.size 1 debezium { database.oracle.jdbc.timezoneAsRegion false } } }6.6 从指定时间戳启动使用startup.mode timestamp从 Unix 毫秒时间戳换算出的 Oracle SCN 开始同步source { Oracle-CDC { plugin_output customers username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB startup.mode timestamp startup.timestamp 1700000000000 server-time-zone UTC debezium { database.oracle.jdbc.timezoneAsRegion false } } }6.7 配置 Debezium 心跳Heartbeat低流量场景下Oracle LogMiner 的 SCN 只有在 redo log 发生变化时才会前进。配置 Debezium 心跳可以让 SCN 持续移动使 checkpoint 位点被定期记录、复制延迟保持可观测。心跳表必须在任务启动前已在 Oracle 服务端存在source { Oracle-CDC { username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] table-names [ORCLCDB.DEBEZIUM.FULL_TYPES] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB debezium { database.oracle.jdbc.timezoneAsRegion false heartbeat.interval.ms 100 heartbeat.action.query INSERT INTO DEBEZIUM.heartbeat (ts) VALUES (SYSTIMESTAMP) } } }6.8 读取无主键表选择与源表保证程度匹配的路径仅追加Append-only负载下游永不产生 UPDATE/DELETE保持exactly_once false且不声明主键。源会退化为尽力而为的行标识没有可用键时连接器无法安全地应用 UPDATE/DELETE 事件。存在唯一非主键列通过table-names-config.primaryKeys声明该列并设置exactly_once true使快照阶段与 redo log 阶段使用同一把键保持行标识一致。source { Oracle-CDC { username system password top_secret database-names [ORCLCDB] schema-names [DEBEZIUM] url jdbc:oracle:thin://oracle-host:1521/ORCLCDB table-names [ORCLCDB.DEBEZIUM.FULL_TYPES_NO_PRIMARY_KEY] table-names-config [ { table ORCLCDB.DEBEZIUM.FULL_TYPES_NO_PRIMARY_KEY primaryKeys [ID] } ] exactly_once true } }没有可用主键时连接器无法安全地应用 UPDATE/DELETE 事件该模式只应用于仅追加负载。6.9 Schema 变更事件过滤当schema-changes.enabled true时可用schema-changes.include/schema-changes.exclude进一步控制哪些变更事件向下游传播。使用 SeaTunnel 的规范事件名规范名操作add.column加列drop.column删列modify.column修改列类型/属性列名不变change.column重命名列可选地改类型update.columns上述四种列级变更的组别名优先级是确定性的① 若设置了schema-changes.include只有被包含的事件类型有资格下发② 随后应用schema-changes.exclude③ 同一类型同时出现在两个列表时exclude 胜出。source { Oracle-CDC { # ... schema-changes.enabled true schema-changes.include [add.column, drop.column] schema-changes.exclude [change.column] } }关于排除drop.column的数据处理注意对于被保留的NOT NULL列NULL写入会被 sink 拒绝——如果对某个源端已停止供数的 NOT NULL 列排除drop.column会在 sink 端失败。6.10 使用 Debezium 兼容格式发送到 Kafkaformat COMPATIBLE_DEBEZIUM_JSON必须与 Kafka sink 连接器配合使用细节参见 cdc-compatible-debezium-json 格式文档。七、源码级实现解析结合 connector-cdc-oracle 模块可以进一步理解文档中各项行为的底层支撑1. LogMiner 挖掘策略与只读模式。OracleSourceConfigFactory.create 会向 Debezium 属性集中写入若干 SeaTunnel 内部约定log.mining.read.only true只读挖掘模式对应 changelog 中的 ReadOnlyLogWriterFlushStrategy 支持tombstones.on.delete false禁用墓碑消息默认log.mining.strategy online_catalog而当schema-changes.enabled true时切换为redo_log_catalog——因为online_catalog无法正确捕获 DDL。源码中还有一处显式防御若用户在debezium块中把include.schema.changes设为 true 但同时指定了online_catalog策略会直接抛出 IllegalArgumentException。2. RedoLogOffset 位点结构。RedoLogOffset 用scn、commit_scn、lcr_position三个键描述 redo log 事件位点compareTo通过 Debezium 的Scn值做顺序比较NO_STOPPING_OFFSETLong.MIN_VALUE表示永不停止对应stop.mode never的流式语义。3. 快照切分ChunkSplitter。OracleChunkSplitter 实现了queryMinMax、sampleDataFromColumn调用OracleUtils.skipReadAndSortSampleData做跳读采样、queryNextChunkMax迭代式不等大分片与queryApproximateRowCnt等方法正是第四节中chunk-key.even-distribution.factor.*、inverse-sampling.rate、split.allow-sampling等参数的执行落点对ROWID类型切分键还使用了ROWID.compareBytes做字节级比较。enable_concurrent_read false时该切分分析整体被跳过、整表作为单 split 读取。4. Schema 变更解析。Oracle 的 DDL 变更由 OracleSchemaChangeResolver 委托给基于 ANTLR 的 CustomOracleAntlrDdlParser 解析 redo log 中的 DDL 语句产出AlterTableColumnEvent事件序列——这解释了为什么当前仅支持加列、删列、改名、改类型这四类列级变更与 OracleIncrementalSource.supports 中声明的ADD_COLUMN / DROP_COLUMN / RENAME_COLUMN / UPDATE_COLUMN一致。5. 任务分发。OracleDialect.createFetchTask 依据 split 类型分发到OracleSnapshotFetchTask快照阶段或OracleRedoLogFetchTask增量阶段内部复用 Debezium 的 LogMiner streaming event source 与只读日志写入冲刷策略 ReadOnlyLogWriterFlushStrategy。八、FAQQ1CDC 需要哪些 Oracle 权限LogMiner 用户需要以下权限GRANT CREATE SESSION TO logminer_user; GRANT SET CONTAINER TO logminer_user; GRANT SELECT ON V_$DATABASE TO logminer_user; GRANT FLASHBACK ANY TABLE TO logminer_user; GRANT SELECT ANY TABLE TO logminer_user; GRANT SELECT_CATALOG_ROLE TO logminer_user; GRANT EXECUTE_CATALOG_ROLE TO logminer_user; GRANT SELECT ANY TRANSACTION TO logminer_user; GRANT LOGMINING TO logminer_user; GRANT CREATE TABLE TO logminer_user; GRANT LOCK ANY TABLE TO logminer_user; GRANT CREATE SEQUENCE TO logminer_user; GRANT EXECUTE ON DBMS_LOGMNR TO logminer_user; GRANT EXECUTE ON DBMS_LOGMNR_D TO logminer_user; GRANT SELECT ON V_$LOG TO logminer_user; GRANT SELECT ON V_$LOG_HISTORY TO logminer_user; GRANT SELECT ON V_$LOGMNR_LOGS TO logminer_user; GRANT SELECT ON V_$LOGMNR_CONTENTS TO logminer_user; GRANT SELECT ON V_$LOGMNR_PARAMETERS TO logminer_user; GRANT SELECT ON V_$LOGFILE TO logminer_user; GRANT SELECT ON V_$ARCHIVED_LOG TO logminer_user; GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO logminer_user; GRANT SELECT ON V_$TRANSACTION TO logminer_user;同时在数据库级和表级启用补充日志ALTER DATABASE ADD SUPPLEMENTAL LOG DATA; ALTER TABLE schema_name.table_name ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;Q2支持多租户CDB/PDB数据库吗支持。将database-names设置为 CDB 名JDBC URL 指向 CDB root用户必须是 common userC##前缀且上述权限以CONTAINER ALL授予所有容器。Q3支持无主键表吗默认要求主键。若表中存在合适的唯一列可通过table-names-config的primaryKeys字段指定自定义主键列无可用主键时只能用于仅追加负载。Q4如何使用自定义快照查询通过debezium块内的 Debezium 属性snapshot.select.statement.overrides配置。该查询在 SeaTunnel 添加快照切分边界之前应用因此必须包含配置表 schema 及其切分键所需的全部列debezium { snapshot.select.statement.overrides DEBEZIUM.FULL_TYPES snapshot.select.statement.overrides.DEBEZIUM.FULL_TYPES SELECT * FROM DEBEZIUM.FULL_TYPES WHERE ACTIVE 1 }Q5如何提升 LogMiner 性能应将其首先当作数据库与 redo log 调优问题处理优先复用本文的 LogMiner 配置与补充日志小节仅为所需表启用日志在此基础上再考虑通过debezium透传属性调优且须先验证这些属性在你实际部署的 Oracle CDC 运行时版本中受支持。九、延伸阅读覆盖「全量 增量」同步完整生命周期、2PC sink 配置、Schema Evolution 与排障的生产级指南CDC Production Cookbook兼容 Debezium JSON 格式用于 Kafka sinkcdc-compatible-debezium-jsonOracle CDC 连接器变更历史connector-cdc-oracle changelog端到端测试connector-cdc-oracle-e2e【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考