跳到主要内容

Oracle Connector

Oracle CDC Pipeline 连接器允许从 Oracle 数据库读取快照数据和增量数据，并提供端到端的整库数据同步能力。本文描述了如何设置 Oracle CDC Pipeline 连接器。

示例​

从 Oracle 读取数据同步到 Doris 的 Pipeline 可以定义如下：

source:
   type: oracle
   name: Oracle Source
   hostname: 127.0.0.1
   port: 1521
   username: debezium
   password: password
   database: ORCLDB
   tables: testdb.\.*, testdb.user_table_[0-9]+, [app|web].order_\.*

sink:
  type: doris
  name: Doris Sink
  fenodes: 127.0.0.1:8030
  username: root
  password: password

pipeline:
   name: Oracle to Doris Pipeline
   parallelism: 4

连接器配置项​

Option Required Default Type Description
hostname required (none) String Oracle 数据库服务器的 IP 地址或主机名。
port required 1521 Integer Oracle 数据库服务器的整数端口号。
username required (none) String 连接到 Oracle 数据库服务器时要使用的 Oracle 用户的名称。
password required (none) String 连接 Oracle 数据库服务器时使用的密码。
tables required (none) String 需要监视的 Oracle 数据库的表名。表名支持正则表达式，以监视满足正则表达式的多个表。
需要注意的是，点号（.）被视为数据库和表名的分隔符。如果需要在正则表达式中使用点（.）来匹配任何字符，必须使用反斜杠对点进行转义。
例如，db0.\.*, db1.user_table_[0-9]+, db[1-2].[app|web]order_\.*
schema-change.enabled optional true Boolean 是否发送模式更改事件，下游 sink 可以响应模式变更事件实现表结构同步，默认为 true。
scan.incremental.snapshot.chunk.size optional 8096 Integer 表快照的块大小（行数），读取表的快照时，捕获的表被拆分为多个块。
scan.snapshot.fetch.size optional 1024 Integer 读取表快照时每次读取数据的最大条数。
scan.startup.mode optional initial String Oracle CDC 消费者可选的启动模式， 合法的模式为 "initial"、"latest-offset" 和 "snapshot"。
debezium.* optional (none) String 将 Debezium 的属性传递给 Debezium 嵌入式引擎，该引擎用于从 Oracle 服务器捕获数据更改。 例如：'debezium.snapshot.mode' = 'never'。 查看更多关于 Debezium 的 Oracle 连接器属性
scan.incremental.close-idle-reader.enabled optional false Boolean 是否在快照结束后关闭空闲的 Reader。此特性需要 flink 版本大于等于 1.14 并且 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' 需要设置为 true。
若 flink 版本大于等于 1.15，'execution.checkpointing.checkpoints-after-tasks-finish.enabled' 默认值变更为 true，可以不用显式配置 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = true。
metadata.list optional false String 可额外读取的 SourceRecord 中元数据的列表，后续可直接使用在 transform 模块，英文逗号 `,` 分割。目前可用值包含：op_ts。
jdbc.url optional (none) String oracle jdbc 的 url，将优先使用 url，如果没有配置 url，则使用 “jdbc:oracle:sinen:@localhost:1521:orcl”，但 oracle 19c url 是 “jdbc:oracle：thin:@//localhost:1521/pdb1”，因此 url 属性是选项，以适应不同版本的 oracle。
database required (none) String 要监视的 Oracle 服务器的数据库名称。
server-time-zone optional (none) String 数据库服务器中的会话时区。如果没有设置，则使用 ZoneId.systemDefault（）来确定服务器时区。
connection.pool.size optional 20 Integer 连接池大小。
connect.timeout optional 30 Integer 连接器在尝试连接到 oracle 数据库服务器后超时前应等待的最长时间。
connect.max-retries optional 3 Integer 连接器应重试构建 oracle 数据库服务器连接的最大重试次数。
chunk-meta.group.size optional 1000 Integer 块元的组大小，如果元大小超过组大小，则元将被分为多个组。
chunk-key.even-distribution.factor.upper-bound optional 1000.0d Double 块密钥分布因子的上限。分布因子用于确定“ 桌子是否均匀分布。" 当数据分布均匀时，表块将使用均匀计算优化。” 当查询 oracle 不均匀时，就会发生拆分。 分布因子可以通过（MAX（id）-MIN（id）+1）/rowCount 来计算。
chunk-key.even-distribution.factor.lower-bound optional 0.05d Double 块密钥分布因子的下限。分布因子用于确定 桌子是否均匀分布。" 当数据分布均匀时，表块将使用均匀计算优化。” 当查询 oracle 不均匀时，就会发生拆分。 分布因子可以通过（MAX（id）-MIN（id）+1）/rowCount 来计算。
debezium.log.mining.strategy optional online_catalog String 日志数据分析或挖掘中的一种策略。
scan.incremental.snapshot.enabled optional true Boolean 增量快照是一种读取表快照的新机制。“与旧的快照机制相比，增量快照具有许多优点，包括：（1）在快照读取过程中源可以并行，（2）源可以在快照读取期间以块粒度执行检查点，（3）源在快照读取之前不需要获取全局读锁（FLUSH TABLES WITH read lock）。
scan.incremental.snapshot.backfill.skip optional false Boolean 是否在快照读取阶段跳过回填。如果跳过回填，则快照阶段捕获的表上的更改将在稍后的增量读取阶段被使用，而不是合并到快照中。警告：跳过回填可能会导致数据不一致，因为在快照阶段发生的一些 binlog 事件可能会被重放（至少只有一次语义承诺）。例如，更新快照中已更新的值，或删除快照中已删除的条目。这些重放的增量事件应该特别处理。

启动模式​

配置选项 scan.startup.mode 指定 Oracle CDC 使用者的启动模式。有效枚举包括：

  • initial（默认）：在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的归档日志。
  • latest-offset：首次启动时，从不对受监视的数据库表执行快照，连接器仅从归档日志的结尾处开始读取，这意味着连接器只能读取在连接器启动之后的数据更改。
  • snapshot: 只进行快照阶段，跳过增量阶段，快照阶段读取结束后退出。

例如，可以在 YAML 配置文件中这样指定启动模式：

source:
  type: oracle
  scan.startup.mode: earliest-offset                    # Start from earliest offset
  scan.startup.mode: latest-offset                      # Start from latest offset
  scan.startup.mode: snapshot                          # Read snapshot only
  # ...

数据类型映射​

Oracle type CDC type NOTE
NUMBER(p,s) DECIMAL(p, s)/BIGINT 当 s 大于 0 时，使用 DECIMAL(p, s)；否则使用 BIGINT。
LONG
 BIGINT
DATE DATE
FLOAT
BINARY_FLOAT
REAL
 FLOAT
BINARY_DOUBLE
DOUBLE 		DOUBLE
TIMESTAMP(p) TIMESTAMP [(p)]
TIMESTAMP(p) WITH TIME ZONE TIMESTAMP_TZ [(p)]
TIMESTAMP(p) WITH LOCAL TIME ZONE TIMESTAMP_LTZ [(p)]
INTERVAL YEAR(2) TO MONTH
INTERVAL DAY(3) TO SECOND(2)
BIGINT 		BIGINT
NCHAR(n)
CHAR(n)
 CHAR(n)
VARCHAR(n)
VARCHAR2(n)
NVARCHAR2(n)
 VARCHAR(n)
CLOB
NLOB
NCLOB
SDO_GEOMETRY
XMLTYPE 		STRING 目前，对于 Oracle 中的 BLOB 数据类型，仅支持长度不大于 2147483647（2**31-1）的 blob。
BLOB
RAW
BFILE 		BYTES
INTEGER
SMALLINT
TINYINT 		INT
BOOLEAN BOOLEAN

Copyright © 2026 The Apache Software Foundation. Apache Flink, Flink, and the Flink logo are trademarks of The Apache Software Foundation.