搭建DM8至加密认证Kafka同步（dmhs）

1.前言

1.1 同步概述

DMHS支持将源端全量或者增量的表数据以json格式投递给消息队列Kafka，应用于基于Kafka订阅消费的数据同步场景。

Kafka同步总体架构包含源端和目标端；其中源端与常规的关系数据库之间的数据同步保持一致，主要由CPT模块进行增量数据捕获，然后将其投递给网络NET模块，通过TCP/IP网络发送至执行端。 Kafka执行端则与常规的数据库同步架构有所区别，DMHS的执行EXEC模块不再由MGR模块调用，以运行执行程序，而是由Java同步程序进行调用。执行模块获取到同步数据后，转换为json格式串。Java同步程序通过JNI接口调用DMHS内部相关接口获取json串后，发送至Kafka消息队列。

本文对DM8同步至加密认证Kafka消息队列的部署配置进行描述。

1.2环境说明

2.DM8准备工作

2.1启用归档（数据实时同步时需要）

DMHS同步源端数据库需要开启归档模式，使用如下语句检查DM8归档是否开启：

SQL> select arch_mode from v$database;LINEID     ARCH_MODE
----------     ---------
1           Y
used time: 3.236(ms). Execute id is 123.

如果ARCH_MODE字段查询结果为“N”，则DM8数据库归档未开启，使用如下命令开启归档模式。

SQL> alter database mount;
executed successfully
used time: 00:00:01.144. Execute id is 0.
SQL> alter database add archivelog 'DEST=/opt/dm8_arch,TYPE=LOCAL,FILE_SIZE=256,SPACE_LIMIT=0';
executed successfully
used time: 7.437(ms). Execute id is 0.
SQL> alter database archivelog;
executed successfully
used time: 10.277(ms). Execute id is 0.
SQL> alter database open;
executed successfully
used time: 00:00:01.262. Execute id is 0.

注意：

在数据库运行过程中，若需要开启归档模式，需要与DBA进行沟通。

2.2启用附加日志（数据实时同步时需要）

sp_set_para_value(2, 'RLOG_APPEND_LOGIC', 1);

注意：

(1) 该参数取值说明如下：

取值为1：如果有主键列，记录UPDATE和DELETE操作时只包含主键列信息；若没有主键列则包含所有列信息；

取值为2：不论是否有主键列，记录UPDATE和DELETE操作时都包含所有列的信息；

取值为3：：记录UPDATE时包含更新列的信息以及ROWID，记录DELETE时只有ROWID。

在Kafka执行端，对于主键表，如果在更新和删除时需要显示全列信息，则需要设置该值为2，保证源端日志中记录有全列信息，否则无法显示全列。

(2)启用该参数后，源端DM8的归档日志量比不启用该参数会有所增加。

2.3创建同步辅助表及触发器（数据实时同步时需要）

执行位于DMHS安装目录的scripts子目录下，命名为：ddl_sql_dm8.sql脚本

3.DMHS源端准备工作

源端安装dmhs，DMHS安装及服务启动脚本可参考如下链接

https://blog.csdn.net/quorra111/article/details/122046715

4.Kafka执行端准备工作 

4.1 DMHS部署安装

Kafka执行端的安装与源端的安装类似。需要注意的是Kafka执行端还存在如下相关组件或脚本：

4.2 JDK环境检查及安装 

Kafka执行端需要启动Java同步程序，因此执行端需要安装Java JDK环境。

4.3配置启动脚本

Kafka启动脚本为DmhsServiceexec_kafka，以下内容对比源端dmhs启动脚本，修改相关地方即可：

#set execute environment
#REPLACE DMHS_HOME path
DMHS_HOME=/opt/dmhs/INST_HS_KFEXEC5
#REPLACE program dir
PROG_DIR=/opt/dmhs/INST_HS_KFEXEC5
#REPLACE program config path
CONF_PATH=/opt/dmhs/INST_HS_KFE