Debezium发布历史129

原文地址： https://debezium.io/blog/2022/10/06/debezium-oracle-series-part-2/

欢迎关注留言，我是收集整理小能手，工具翻译，仅供参考，笔芯笔芯.

Debezium for Oracle - Part 2: Running the connector
October 6, 2022 by Chris Cranford
debezium oracle examples

这篇文章是一个由三部分组成的系列文章的一部分,该系列文章旨在探讨使用Debezum来吸收甲骨文数据库中使用甲骨文日志程序的更改。如果你错过了,本系列的第一部分是在这里 .

在第二部分中,我们将在第一部分的基础上,使用动物园管理员、卡夫卡和卡夫卡连接来部署甲骨文连接器。我们将讨论连接器的各种配置选项,以及它们为什么重要。最后,我们将看到连接器的运行!

建立卡夫卡连接和先决条件
为了使用Debezum,需要启动三个单独的服务:

zookeeper

kafka broker

kafka connect

我们会利用码头工人运行上述服务的集装箱。使用单独的容器简化了部署过程,这样您就可以看到Debez铵在起作用。此外,我们亦会下载甲骨文JDBC驱动程序并将其挂载为卡夫卡连接容器的一部分。

在生产中使用这些服务的多个实例提供性能、可靠性和容错性。部署通常会涉及一个平台,比如用于管理多个容器的开放转移或库伯内特斯,或者您将使用专用硬件并手动管理这个。

对于这个博客,我们将使用每个服务的单个实例来保持它的简单性。

动物园管理员和卡夫卡容器是短暂的。通常情况下,卷将被安装在主机机器上,以便在容器停止时,由容器管理的数据会继续存在。为了简单起见,我们跳过了这一步,这样当容器停止时,数据将丢失。

先决条件:动物园管理员
动物园管理员服务是第一个开始的服务。卡夫卡经纪人使用动物园管理员处理卡夫卡经纪人的领导权选举,并管理集群中的服务发现,以便每个经纪人知道当一个经纪人终止时,一个兄弟姐妹何时加入或离开,以及新的领导人是为某个特定主题/分区元组服务的。

打开一个新的终端窗口并运行以下命令:

docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888
quay.io/debezium/zookeeper:1.9
… 饲养员集装箱以交互方式启动,并在停止时销毁。容器被命名为zookeeper ,这在启动未来集装箱时非常重要。

先决条件:开始卡夫卡
卡夫卡服务是必须启动的第二项服务,它取决于动物园管理员的服务。Debezum生成更改事件发送到由卡夫卡经纪人管理的主题。

打开一个新的终端窗口并运行以下命令:

docker run -it --rm --name kafka -p 9092:9092 --link zookeeper:zookeeper
quay.io/debezium/kafka:1.9
… 卡夫卡集装箱以交互方式启动,并在停止时销毁。容器被命名为kafka ,这将是重要的启动未来的容器。此外, 卡夫卡服务还链接到饲养员服务,意思是标准名称zookeeper 会对运行饲养员服务。

先决条件:下载甲骨文JDBC驱动程序
德贝兹卡夫卡连接图像不与甲骨文JDBC驱动程序发送。若要对甲骨文使用Debezns,JDBC驱动程序必须手动下载并安装到Debezikafka连接图像中。

航行到 JDBC驱动程序下载页面。在发布时,甲骨文最新的数据库是甲骨文21,所以点击ojdbc8.jar 链接在甲骨文21C部分。下载的JAR将在下一节中使用,将驱动程序添加到Debezns的卡夫卡连接容器的基本图像中。

先决条件:启动卡夫卡连接
卡夫卡连接服务是第三个也是最后一个必须启动的服务,它依赖于卡夫卡服务。卡夫卡连接负责管理所有连接器及其相关工作负载,并且是运行甲骨文的Debezum连接器的运行时的运行时环境。

打开一个新的终端窗口并运行以下命令:

docker run -it --rm --name connect -p 8083:8083
-e GROUP_ID=1
-e CONFIG_STORAGE_TOPIC=my_connect_configs
-e OFFSET_STORAGE_TOPIC=my_connect_offsets
-e STATUS_STORAGE_TOPIC=my_connect_statuses
–link kafka:kafka
–link dbz_oracle21:dbz_oracle21
-v /path/to/ojdbc8.jar:/kafka/libs/ojdbc8.jar
quay.io/debezium/connect:1.9
… 连接集装箱以交互方式启动,并在停止时销毁。容器被命名为connect ,几个环境变量控制几个需要的主题和一些需要的配置参数的命名。此外, 连接集装箱连结至卡夫卡容器,意思是标准名称kafka 会对运行卡夫卡经纪服务。

与先前的容器不同,我们使用-v 指挥官。论点采用了local-path:container-path .

… 当地路径代表了ojdbc8.jar 文件存在于主机上.… 集装箱式通道应保留/kafka/libs/ojdbc8.jar ,在卡夫卡连接类路径上安装驱动程序。

创建一些初始测试数据
如果在本系列的第一部分中创建的甲骨文数据库使用甲骨文容器注册表映像,则数据库中不存在初始数据。虽然这并不一定会带来问题,但理想的情况下,我们希望在部署甲骨文连接器时快照一些数据;因此,在部署之前必须存在一些初始数据。

在一个新的终端中,让我们使用sql+连接到数据库,并创建一个带有一些初始数据的新表。下面使用普通用户,连接到可插入数据库ORCLPDB1 .在连接到具有要捕获的表的现有环境时,可以安全地跳过这一步。

docker exec -it -e ORACLE_SID=ORCLPDB1 dbz_oracle21 sqlplus c##dbzuser@ORCLPDB1
连接后,使用以下SQL创建表和一些初始数据:

CREATE TABLE customers (id number(9,0) primary key, name varchar2(50));
INSERT INTO customers VALUES (1001, ‘Salles Thomas’);
INSERT INTO customers VALUES (1002, ‘George Bailey’);
INSERT INTO customers VALUES (1003, ‘Edward Walker’);
INSERT INTO customers VALUES (1004, ‘Anne Kretchmar’);
COMMIT;
默认情况下,重做日志只捕获最小的更改信息。CUSTOMERS 因为补充日志记录只设置在数据库级别。

如果你熟悉后格来格的REPLICA IDENTITY 或者我的binlog_format 甲骨文提供了一个类似的机制,称为表级补充日志记录,我们在本系列的第一部分中提到了它。表级别的补充日志控制用户修改行时重做日志中捕获的列。将表的补充日志级别设置为(ALL) COLUMNS 甲骨文捕捉到与之相关的变化的保证INSERT ,UPDATE ,以及DELETE 重做日志中的操作。

使用以下SQL设置表的补充日志级别:

ALTER TABLE customers ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
假设捕获的表的补充日志级别设置错误。在这种情况下,连接器将记录一个警告,让您知道存在一个问题,这样您就可以调整表的设置来捕获所有更改。

值得注意的是,虽然这个示例使用相同的用户帐户来创建这个。CUSTOMERS 连接器用来连接的表,连接器使用的用户与甲骨文数据库中拥有表的用户有所不同,这一点并不少见。在这种情况下,连接器用户必须拥有读取捕获的表的权限,要求SELECT 每桌允许。

部署甲骨文连接器
我们现在准备好了,准备好了。在注册卡夫卡连接连接器之前,让我们深入研究一下配置。

下面是我们将在这个示例中使用的示例配置:

{
“name”: “customers-connector”,
“config”: {
“connector.class”: “io.debezium.connector.oracle.OracleConnector”,
“tasks.max”: “1”,
“database.hostname”: “dbz_oracle21”,
“database.port”: “1521”,
“database.user”: “c##dbzuser”,
“database.password”: “dbz”,
“database.dbname”: “ORCLCDB”,
“database.pdb.name”: “ORCLPDB1”,
“database.server.name”: “server1”,
“table.include.list”: “C##DBZUSER.CUSTOMERS”,
“database.history.kafka.bootstrap.servers”: “kafka:9092”,
“database.history.kafka.topic”: “schema-changes”
}
}
让我们来看看这些配置选项的含义。

name
这是分配给连接器的名称,在卡夫卡连接集群中,连接器必须是唯一的。

connector.class
这是已部署连接器的类实现。每一个Debezum源连接器都有一个唯一的类名称来标识正在部署的连接器。

tasks.max
这是分配给卡夫卡连接中连接器部署的最大任务数。大多数德贝兹连接器依次从源数据库读取昌塞,因此,一个值1 很有道理。

database.hostname
这是数据库主机名或IP地址。因为我们指定了一个链接dbz_oracle21 当启动卡夫卡连接时,我们可以在这里使用这个名称来标识运行甲骨文数据库的容器。如果在另一个主机上有一个原有的甲骨文环境,请在此配置属性中指定该主机的名称。

database.port
这是数据库用来监听连接的端口。甲骨文的默认端口是1521 但是数据库管理员可以将其配置为任何可用端口。如果您正在连接到先前的甲骨文实例,请使用数据库使用的端口。

database.user
这是用于JDBC连接的数据库用户帐户。这应该是本系列第一部分中创建的通用用户,即c##dbzuser 用户。如果您正在连接到一个不支持多租户的环境,那么这将是您在没有普通用户前缀的根数据库中创建的用户。

database.password
这是数据库用户帐户密码.

database.dbname
这是连接器通信的数据库服务。不管是否启用了多租户,这总是单数或根容器数据库。

database.pdb.name
这是可选的可插入数据库系统标识符。在连接到支持多租户并引用PDB的数据库时,必须提供此属性。如果省略此字段,连接器假设数据库不支持多租户。

database.server.name
连接器创建的所有主题的前缀。在卡夫卡连接集群中的所有主题部署中,这个值必须是唯一的。

table.include.list
用逗号分隔的正则表达式或简单的表名称列表.

识别连接器将捕获的表。

database.history.kafka.bootstrap.servers
这是存储数据库历史主题的卡夫卡代理的URL。因为我们指定了一个链接kafka 当启动卡夫卡连接时,我们可以在这里使用这个名称来指向代理及其端口。

database.history.kafka.topic
这是将存储数据库架构历史的主题名称。连接器重新启动时将恢复此主题,从此主题填充内存中的关系模型。

除了后GERGSQL之外,所有的Debezum连接器都使用架构历史来存储所有表的模式。这通常不适合甲骨文数据库,特别是在部署没有多租赁的连接器时。

若要将存储限制为仅包含列表中的表,请通过设置database.history.store.only.captured.tables.ddl 财产true .

有关其他连接器属性的更多信息,您可以查看甲骨文文件更多的细节。

若要部署连接器,请将上述配置保存到一个称为register-oracle.json .现在,打开一个新的终端窗口curl 命令用卡夫卡连接注册连接器:

curl -i -X POST -H “Accept:application/json”
-H “Content-Type:application/json”
localhost:8083/connectors
-d @register-oracle.json
如注册成功,则由连接容器正在运行,将开始执行数据在CUSTOMERS 表我们还可以通过使用卡夫卡控制台消费者工具和将主题的内容读取到本地终端来确认卡夫卡中存在的数据。

要检查主题的内容,请使用注册连接器的相同终端并执行以下命令:

docker exec -it kafka /kafka/bin/kafka-console-consumer.sh
–bootstrap-server 0.0.0.0:9092
–from-beginning
–property print.key=true
–topic server1.C__DBZUSER.CUSTOMERS
主题将架构名称从C##DBZUSER 到C__DBZUSER 因为主题命名策略自动保证主题的名称与avro兼容,后者不允许哈希符号字符。

上述命令的输出应类似于以下内容:

{
“schema”:{
…
},
“payload”:{
“before”:null,
“after”:{
“ID”:“1001”,
“NAME”:“Salles Thomas”
},
“source”:{
“version”:“1.9.6.Final”,
“connector”:“oracle”,
“name”:“server1”,
“ts_ms”:1665102121000,
“snapshot”:“true”,
“db”:“ORCLPDB1”,
“sequence”:null,
“schema”:“C##DBZUSER”,
“table”:“CUSTOMERS”,
“txId”:null,
“scn”:“2868546”,
“commit_scn”:null,
“lcr_position”:null,
“rs_id”:null,
“ssn”:0,
“redo_thread”:null
},
“op”:“r”,
“ts_ms”:1665102126961,
“transaction”:null
}
}
…
现在您可以使用sql+终端,在那里您创建了初始测试数据。INSERT ,UPDATE ,或DELETE 的纪录CUSTOMERS 表您将在终端看到相应的更改事件,该事件正在跟踪server1.C__DBZUSER.CUSTOMERS 专题。