Debezium Oracle Source Connector

Debezium Oracle Source Connector

Debezium Oracle Source Connector 是一个可以连接器捕获并记录 Oracle 服务器上的数据库中发生的行级更改，包括连接器运行时添加的表。您可以将连接器配置为针对架构和表的特定子集发出更改事件，或者忽略、屏蔽或截断特定列中的值。
有关与此连接器兼容的 Oracle 数据库版本的信息，请参阅Debezium 版本概述。
Debezium 通过使用本机 LogMiner 数据库包、XStream API 或 OpenLogReplicator 从 Oracle 获取更改事件。

• 支持全量快照，在第一次启动时抽取监视表的全量数据；
• 支持增量快照，分块抽取全量数据，并与数据变化事件交错传递到下游，适合表中有海量数据的场景；
• 支持即席快照，即手动触发全量快照或增量快照，可用数据补录等；
• 全量迁移和数据同步无缝衔接，完成全量快照迁移后，自动启动增量同步，即使迁移过程中发数据变化也不会丢失数据；
• 高可用部署，依靠kafka connect集群特性，如若遇到故障自动重启；
• 正好一次语义，所有状态数据记录在特定的kafka topic中，即使重启也不会导致数据丢失或重复；
• 支持事务边界事件，支持在事件流中添加事务开始和结束事件，以及在每个事件中标识事务ID，从而在下游应用可以以事务为单位处理事件流；
• 支持捕获DDL事件；

一、安装 Debezium Oracle Connector

你可以手动下载压缩文件来安装此连接器，也可以通过Linux命令来安装。

方式1：手动安装连接器

手动下载并解压连接器的压缩文件，然后按照说明进行操作。

https://debezium.io/releases/

https://debezium.io/releases/2.4/#releases

上传 /usr/local/src 目标并解压

cd /usr/local/src && tar -zxvf debezium-connector-oracle-2.4.2.Final-plugin.tar.gz -C  /home/kafka/plugins

方式2：命令安装连接器

通过Linux命令下载并解压连接器的压缩文件，然后按照说明进行操作。

• https://repo1.maven.org/maven2/io/debezium/debezium-connector-oracle

上传 /usr/local/src 目标并解压

$ cd /usr/local/src && wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-oracle/2.4.1.Final/debezium-connector-oracle-2.4.2.Final-plugin.tar.gz --no-check-certificate
$ tar -zxvf debezium-connector-oracle-2.4.2.Final-plugin.tar.gz -C  /home/kafka/plugins

二、开启 Oracle 数据库 binlog 功能

log-bin=mysql-bin
binlog_format=row
server-id = 1
binlog_row_image=FULL

三、创建 Oracle Connector 监听

https://debezium.io/documentation/reference/2.4/connectors/oracle.html

{
    "name": "inventory-connector",
    "config": {
        "connector.class" : "io.debezium.connector.oracle.OracleConnector",
        "database.hostname" : "<ORACLE_IP_ADDRESS>",
        "database.port" : "1521",
        "database.user" : "c##dbzuser",
        "database.password" : "dbz",
        "database.dbname" : "ORCLCDB",
        "topic.prefix" : "server1",
        "tasks.max" : "1",
        "database.pdb.name" : "ORCLPDB1",
        "schema.history.internal.kafka.bootstrap.servers" : "kafka:9092",
        "schema.history.internal.kafka.topic": "schema-changes.inventory"
    }
}

参考资料

https://blog.csdn.net/fayeestone/article/details/120839873
https://huaweicloud.csdn.net/637ef519df016f70ae4ca5f5.html