热门搜索
分布式
数据库一体机
云数据库
建设银行
保险核心
密集交易
注册
Linux系统配置DRS达梦-kafka同步
培训园地/ 文章详情 /

Linux系统配置DRS达梦-kafka同步

pink。 2026/08/17 243 0 0

1、环境规划

1.1数据库规划

image.png

1.2DRS规划

image.png

1.3端口规划

image.png

2、部署kafka

2.1安装JDK

Kafka需要java运行环境，需要安装JDK。

上传JDK安装包，并安装：
[root@x86_1_59 ~]# tar -zxvf jdk-8u60-linux-x64.tar.gz
[root@x86_1_59 ~]# mv jdk1.8.0_60/ /usr/local/src/java
修改环境变量：
[root@x86_1_59 ~]# vim /etc/profile
添加：
export JAVA_HOME=/usr/local/src/java/
export PATH=$JAVA_HOME/bin:$PATH
[root@x86_1_59 ~]# source /etc/profile
查看java环境：
[root@x86_1_59 ~]# java -version

2.2安装kafka

1、上传kafka安装包到home目录，并解压：
如需使用其他版本kafka，可通过http://kafka.apache.org/downloads下载
[root@x86_1_59 home]# tar -zxvf kafka_2.12-3.1.2.tgz
2、修改server.properties文件
[root@x86_1_59 home]# cd kafka_2.12-3.1.2/config/
[root@x86_1_59 config]# ls
image.png
创建日志存放路径：
[root@x86_1_59 ~]# cd /home/kafka_2.12-3.1.2
[root@x86_1_59 kafka_2.12-3.1.2]# mkdir -p logs/server
[root@x86_1_59 config]# vim server.properties

修改如下内容：
broker.id=0 
port=9092  #端口号 
host.name=10.15.1.59  #单机可直接用localhost 
log.dirs=/home/kafka_2.12-3.1.2/logs/server  #日志存放路径可修改可不修改 
zookeeper.connect=10.15.1.59:2181  #zookeeper地址和端口，单机配置部署，localhost:2181
#因为本次使用的kafka版本中自带zookeeper，使用默认配置即可

3、修改zookeeper.properties文件
创建日志存放路径：
[root@x86_1_59 ~]# cd /home/kafka_2.12-3.1.2
[root@x86_1_59 kafka_2.12-3.1.2]# mkdir -p logs/zookeeper
[root@x86_1_59 config]# vim zookeeper.properties

修改如下内容：
dataDir=/home/zerone/zxf/local/kafka/data/zookeeper #zookeeper数据目录 (可以修改可以不修改）
dataLogDir=/home/zerone/zxf/local/kafka/logs/zookeeper #zookeeper日志目录 （可以修改可以不修改） 
clientPort=2181 
maxClientCnxns=100 
tickTime=2000 
initLimit=10

4、启动zookeeper和kafka
[root@x86_1_59 ~]# cd /home/kafka_2.12-3.1.2/bin/
[root@x86_1_59 bin]# vim kafkaStart.sh

添加：
#!/bin/bash
#启动zookeeper
nohup /home/kafka_2.12-3.1.2/bin/zookeeper-server-start.sh /home/kafka_2.12-3.1.2/config/zookeeper.properties 1>& /home/kafka_2.12-3.1.2/logs/zookeeper/zookeeper.log &
sleep 3  #默默等3秒后执行 
#启动kafka
nohup /home/kafka_2.12-3.1.2/bin/kafka-server-start.sh /home/kafka_2.12-3.1.2/config/server.properties 1>& /home/kafka_2.12-3.1.2/logs/server/kafka.log &

授予脚本执行权限：
[root@x86_1_59 bin]# chmod 755 kafkaStart.sh
执行启动脚本：
[root@x86_1_59 bin]# ./kafkaStart.sh
5、创建topic
[root@x86_1_59 bin]# ./kafka-topics.sh --bootstrap-server 10.15.1.59:9092 --create --replication-factor 1 --partitions 1 --topic test
创建成功后查询top主题
[root@x86_1_59 bin]# ./kafka-topics.sh --list --bootstrap-server 10.15.1.59:9092
image.png
6、测试生成者和消费者
[root@x86_1_59 bin]# ./kafka-console-producer.sh --broker-list 10.15.1.59:9092 --topic test
输入 一些信息，然后启动消费者查询是否可查看到生产者发送的信息
image.png
[root@x86_1_59 bin]# ./kafka-console-consumer.sh --bootstrap-server 10.15.1.59:9092 --topic test --from-beginning
image.png

3、配置达梦端数据库

3.1源端安装DRS

Extract install files.........

Please select the installer's language (E/e:English C/c:Chinese)[E/e]:c
-----------欢迎使用达梦数据复制软件安装工具-----------
输入[exit]可退出安装。
-----------安装目录-----------
指定安装目录[/home/dmdba/dmdrs5]:/home/dmdba/dmdrs5
-----------安装组件-----------
请选择需要安装的组件
1.安装达梦数据融合管理平台
2.安装代理
请选择安装组件数字序号(使用空格间隔):1
选择的组件有:
1.安装达梦数据融合管理平台
"确认?[Y/y(确认选择) or N/n(重新选择)]:y
-----------许可证文件-----------
1.免费试用达梦数据复制软件(必须在试用期范围内使用，反复安装无效，使用时间为3个月)
2.使用许可证文件
指定许可证文件(1,2)[1]:1
免费试用许可证文件限制信息如下：
有效日期:2024-07-11
用户名称:DEVELOP USER
版本编号:4.0
许可证编号:dm66n367
版本:试用版
版本类型:试用版
用户编号:1
并发用户编号:0
数据源类型:DM6,DM7,DM8,ORACLE,SQL SERVER,MYSQL,DB2,PG,KAFKA
DRS主备:支持
动态对比:支持
当前默认许可证文件支持数据库类型为[DM6,DM7,DM8,ORACLE,SQL SERVER,MYSQL,DB2,PG,KAFKA]，是否继续安装?[Y/y or N/n]:y
-----------配置-----------
依赖环境配置 -数据库动态库路径[/cs/dmdbms/bin]:/cs/dmdbms/bin
达梦数据融合管理平台配置-管理平台端口[8080]:8080
是否使用外置库(0:不使用  1:使用 )[0]:0
-----------安装小结-----------
安装目录:[/home/dmdba/dmdrs5]
依赖环境配置-NEED_LIB_PATH:[/cs/dmdbms/bin]
达梦数据融合管理平台IP:[127.0.0.1]
达梦数据融合管理平台端口:[8080]
内置库信息：
数据库IP:[127.0.0.1]
数据库端口:[15236]
用户名:[SYSDBA]
密码:[******]
所需磁盘空间/可用磁盘空间:[1,836 MB/2,217,278 MB]
确认安装?[Y/y or N/n]:y
-----------安装中-----------
server start ...    server finished.
default start ...    default finished.
web start ...    web finished.
doc start ...    doc finished.
db start ...    db finished.
安装成功
-----------系统服务-----------
内置数据库服务设置
1.自动(注册为系统服务)
2.手动(脚本启动)
启动方式(1,2)[2]:
正在创建内置数据库服务....
达梦数据融合管理平台服务设置
1.自动(注册为系统服务)
2.手动(脚本启动)
启动方式(1,2)[2]:
正在创建达梦数据融合管理平台服务....
正在启动服务[DmServiceDFDB]
正在启动服务[DfmWebService]
-----------安装总结-----------
达梦数据复制软件V5安装完成
地址:http://127.0.0.1:8080
用户名/密码:admin/admin
更多安装信息，请查看安装日志文件：/home/dmdba/dmdrs5/log/install.log

3.2安装达梦数据库

略

3.3初始化实例

1、初始化实例

[dmdba@x86_1_55 ~]$ cd /cs/dmdbms/bin
[dmdba@x86_1_55 bin]$ ./dminit PATH=/home/dmdba/data EXTENT_SIZE=32 PAGE_SIZE=32 LOG_SIZE=2048 PORT_NUM=5233

2、注册服务

[root@x86_1_55 ~]# cd /cs/dmdbms/script/root/
[root@x86_1_55 root]# ./dm_service_installer.sh -t dmserver -p DMSERVER -dm_ini /home/dmdba/data/DAMENG/dm.ini

3.3.1开启归档及逻辑日志

1、开启归档

[dmdba@x86_1_55 ~]$ vim /home/dmdba/data/DAMENG/dm.ini                                                                                    
ARCH_INI = 1                        
[dmdba@x86_1_55 data]$ vim /home/dmdba/data/DAMENG/dmarch.ini  
[ARCHIVE_LOCAL1]
ARCH_TYPE = LOCAL
ARCH_DEST = /home/dmdba/data/arch
ARCH_FILE_SIZE = 512
ARCH_SPACE_LIMIT = 10240

2、开启逻辑日志

[dmdba@x86_1_55 ~]$ vim /home/dmdba/data/DAMENG/dm.ini                                       
#redo log                                               
RLOG_APPEND_LOGIC = 1

3、重启数据库生效

[dmdba@x86_1_55 data]$ cd /cs/dmdbms/bin
[dmdba@x86_1_55 bin]$ ./DmServiceDMSERVER restart

3.3.2同步用户

1、创建数据库同步用户。

[dmdba@x86_1_55 bin]$ ./disql sysdba/SYSDBA:5233
SQL> create tablespace DRS datafile '/home/dmdba/data/DAMENG/DRS.DBF' size 2048 autoextend on next 2048;
SQL> CREATE USER DRS IDENTIFIED BY 123456789 default tablespace DRS default index tablespace DRS;

2、赋予同步用户权限
本次测试授予同步用户DBA权限。

GRANT dba TO DRS;

3.3.3配置辅助表

在源数据库执行“ddl_sql_dm8.sql”脚本创建辅助表，脚本默认位置在DRS执行程序目录下的scripts子目录中。
1、创建辅助表

start /home/dmdba/dmdrs5/bin/scripts/ddl_sql_dm8.sql

2、检查创建的辅助表和触发器是否有效

select owner, trigger_name from dba_triggers where owner = 'SYSDBA' and trigger_name like 'DRS_$%' and status = 'Y';

image.png

select owner, table_name from dba_tables where owner = 'SYSDBA' and table_name like 'DRS_$%' and status = 'VALID';

image.png
如果存在以上查询结果，表示DRS辅助表和触发器创建有效。

3.4配置源端DRS配置文件

[dmdba@x86_1_55 ~]$ cd /home/dmdba/dmdrs5/bin
[dmdba@x86_1_55 bin]$ vim cpt.xml
<?xml version="1.0" encoding="GB2312" standalone="no"?>
<drs>
  <base>
    <mgr_port>5345</mgr_port>
    <siteid>1</siteid>
  </base>
  <cpt>
    <name>cpt_dm8</name>
    <login>
      <dbtype>dm8</dbtype>
      <server>10.15.1.55</server>
      <user>SYSDBA</user>
      <pwd>SYSDBA</pwd>
      <port>5233</port>
	  <char_code>GB18030</char_code>
	  <ddl_mask>OBJ:OP</ddl_mask>
    </login>
    <send>
      <ip>10.15.1.59</ip>
      <port>5346</port>
      <target_name>exec_kafka</target_name>
      <map>
        <item>DRS.*==DRS.*</item>
      </map>
    </send>
    <load/>
  </cpt>
</drs>

4、配置目的端DRS

4.1目的端安装DRS

[dmdba@x86_1_55 home]$ ./DRS_V5.0.4.1_rev157302_rh6_64_20240411.bin -i
Extract install files.........

Please select the installer's language (E/e:English C/c:Chinese)[E/e]:c
-----------欢迎使用达梦数据复制软件安装工具-----------
输入[exit]可退出安装。
-----------安装目录-----------
指定安装目录[/home/dmdba/dmdrs5]:/home/dmdba/dmdrs5
-----------安装组件-----------
请选择需要安装的组件
1.安装达梦数据融合管理平台
2.安装代理
请选择安装组件数字序号(使用空格间隔):1
选择的组件有:
1.安装达梦数据融合管理平台
"确认?[Y/y(确认选择) or N/n(重新选择)]:y
-----------许可证文件-----------
1.免费试用达梦数据复制软件(必须在试用期范围内使用，反复安装无效，使用时间为3个月)
2.使用许可证文件
指定许可证文件(1,2)[1]:1
免费试用许可证文件限制信息如下：
有效日期:2024-07-11
用户名称:DEVELOP USER
版本编号:4.0
许可证编号:dm66n367
版本:试用版
版本类型:试用版
用户编号:1
并发用户编号:0
数据源类型:DM6,DM7,DM8,ORACLE,SQL SERVER,MYSQL,DB2,PG,KAFKA
DRS主备:支持
动态对比:支持
当前默认许可证文件支持数据库类型为[DM6,DM7,DM8,ORACLE,SQL SERVER,MYSQL,DB2,PG,KAFKA]，是否继续安装?[Y/y or N/n]:y
-----------配置-----------
依赖环境配置 -数据库动态库路径[/cs/dmdbms/bin]:/cs/dmdbms/bin
达梦数据融合管理平台配置-管理平台端口[8080]:8080
是否使用外置库(0:不使用  1:使用 )[0]:0
-----------安装小结-----------
安装目录:[/home/dmdba/dmdrs5]
依赖环境配置-NEED_LIB_PATH:[/cs/dmdbms/bin]
达梦数据融合管理平台IP:[127.0.0.1]
达梦数据融合管理平台端口:[8080]
内置库信息：
数据库IP:[127.0.0.1]
数据库端口:[15236]
用户名:[SYSDBA]
密码:[******]
所需磁盘空间/可用磁盘空间:[1,836 MB/2,217,278 MB]
确认安装?[Y/y or N/n]:y
-----------安装中-----------
server start ...    server finished.
default start ...    default finished.
web start ...    web finished.
doc start ...    doc finished.
db start ...    db finished.
安装成功
-----------系统服务-----------
内置数据库服务设置
1.自动(注册为系统服务)
2.手动(脚本启动)
启动方式(1,2)[2]:
正在创建内置数据库服务....
达梦数据融合管理平台服务设置
1.自动(注册为系统服务)
2.手动(脚本启动)
启动方式(1,2)[2]:
正在创建达梦数据融合管理平台服务....
正在启动服务[DmServiceDFDB]
正在启动服务[DfmWebService]
-----------安装总结-----------
达梦数据复制软件V5安装完成
地址:http://127.0.0.1:8080
用户名/密码:admin/admin
更多安装信息，请查看安装日志文件：/home/dmdba/dmdrs5/log/install.log

4.2目的端DRS配置json_format.ini文件

[dmdba@x86_1_59 ~]$ cd /home/dmdba/dmdrs5/bin
[dmdba@x86_1_59 bin]$ vim json_format.ini
#DMDRS kafka json format Configuration file
#this is comments
#common format control parameters
OP_TIME_FORMAT					= (yyyy-mm-dd hh:mi:ss)       
CUR_TIME_FORMAT				    = (yyyy-mm-dd hh:mi:ss)	 					
NULL_FORMAT				        = "null"               	
SET_QUOTA						= 0	
SET_ROWID_COL					= 0
NEED_CRLF						= 1				
CHAR_REPLACE					=(",\"),(\,\\)
OLD_VALUES                      = ALL
NEW_VALUES 					    = ALL
OLD_LOB_FLAG                    = "empty_lob()"     
JSON_FORMAT_INS = {
                "table":"#SCHEMA.#TABLE",
                "op_type":"#OP_TYPE",
                "op_ts":"#OP_TIME",
                "current_ts":"#TIME",
                "pos":"#POS",
                "primary_keys":[#PRIMARY_KEY],
                "after":{#NEW_VALUES}
}
JSON_FORMAT_UPD = {
                "table":"#SCHEMA.#TABLE",
                "op_type":"#OP_TYPE",
                "op_ts":"#OP_TIME",
                "current_ts":"#TIME",
                "pos":"#POS",
                "primary_keys":[#PRIMARY_KEY],
                "before":{#OLD_VALUES},
                "after":{#NEW_VALUES}
}
JSON_FORMAT_DEL = {
                "table":"#SCHEMA.#TABLE",
                "op_type":"#OP_TYPE",
                "op_ts":"#OP_TIME",
                "current_ts":"#TIME",
                "pos":"#POS",
                "primary_keys":[#PRIMARY_KEY],
                "before":{#OLD_VALUES}
}
JSON_FORMAT_DDL = {
                "table":"#SCHEMA.#TABLE",
                "op_type":"#OP_TYPE",
                "op_ts":"#OP_TIME",
                "current_ts":"#TIME",
                "pos":"#POS",      
                "after":
                {"DB_TYPE":"ORA",
                "OBJECT_TYPE":"#OP_TYPE",
                "OBJECT_NAME":"#SCHEMA.#TABLE",
                "DDL_TEXT":"#DDL_SQL"}
}

4.3目的端DRS配置producer.properties文件

[dmdba@x86_1_59 ~]$ cd /home/dmdba/dmdrs5/bin
[dmdba@x86_1_59 bin]$ vim producer.properties
bootstrap.servers=10.15.1.59:9092

4.4配置目的端DRS配置文件

[dmdba@x86_1_59 ~]$ cd /home/dmdba/dmdrs5/bin
[dmdba@x86_1_59 bin]$ vim exec.xml
<?xml version="1.0" encoding="GB2312"?>
<drs>
    <base>
        <mgr_port>5346</mgr_port>
        <siteid>1</siteid>
    </base>
    <exec>
        <name>exec_kafka</name>
        <login>
            <dbtype>kafka</dbtype>
            <producer_properties>/home/dmdba/dmdrs5/bin/producer.properties</producer_properties>
        </login>
        <group>
            <item>
                <id>0</id>
                <topic_name>test</topic_name>
                <json_format_path>/home/dmdba/dmdrs5/bin/json_format.ini</json_format_path>
            </item>
        </group>
    </exec>
</drs>

5、配置启动脚本

5.1配置源端启动脚本

[dmdba@x86_1_55 ~]$ cd /home/dmdba/dmdrs5/bin
[dmdba@x86_1_55 bin]$ cp service_template/TemplateService ./DrsServiceCPT
修改DrsServiceCPT服务脚本相关配置
INSTALL_HOME=/home/dmdba/dmdrs5
PROG_DIR=/home/dmdba/dmdrs5/bin
CONF_PATH=/home/dmdba/dmdrs5/bin/cpt.xml
NEED_LIB_PATH=/cs/dmdbms/bin:/home/dmdba/dmdrs5/bin
EXEC_PROG_NAME=drsvr
SERVICE_TYPE_NAME=drs server

5.2配置目的端启动脚本

[dmdba@x86_1_59 ~]$ cd /home/dmdba/dmdrs5/bin
[dmdba@x86_1_59 bin]$ cp service_template/TemplateService ./DrsServiceEXEC
修改DrsServiceCPT服务脚本相关配置
INSTALL_HOME=/home/dmdba/dmdrs5
PROG_DIR=/home/dmdba/dmdrs5/bin
CONF_PATH=/home/dmdba/dmdrs5/bin/exec.xml
NEED_LIB_PATH=/cs/dmdbms/bin:/home/dmdba/dmdrs5/bin
EXEC_PROG_NAME=drsvr
SERVICE_TYPE_NAME=drs server

6、前台方式启动源端和目的端服务，进行同步测试

6.1前台方式启动目的端服务

[dmdba@x86_1_59 ~]$ cd /home/dmdba/dmdrs5/bin
[dmdba@x86_1_59 bin]$ ./drsvr exec.xml

6.2前台方式启动源端服务

[dmdba@x86_1_55 ~]$ cd /home/dmdba/dmdrs5/bin
[dmdba@x86_1_55 bin]$ ./drsvr cpt.xml

6.3开启源端同步

[dmdba@x86_1_55 ~]$ cd /home/dmdba/dmdrs5/bin
[dmdba@x86_1_55 bin]$ ./drcsl cpt.xml 
#连接DRS服务
CSL> connect
#设置数据库当前最新的日志LSN作为日志解析的起始位置
CSL> alter cpt_dm8 set lsn
#启动源DMDRS服务中的CPT模块
CSL> start
#将源数据库中DRS模式下所有的表添加同步，并在同步之前把源端表的全量数据装载到目标数据库。
CSL> alter cpt_dm8 add table "sch.name='DRS'"

image.png

6.4查看源端和目的端日志

1、源端日志
image.png
image.png

2、目的端日志
image.png
此时可通过日志看出同步数据正常

6.5到kafka端验证输出结果

此时消费者输出同步的表和数据，同步正常，无问题。
image.png

image.png

7、过程中的报错

7.1在执行创建topic时报错

Exception in thread "main" joptsimple.UnrecognizedOptionException: zookeeper is not a recognized option
image.png
【解决方式】
原因是kafka版本过高所致，本次测试使用3.1.2版本，2.2+的版本已经不需要依赖zookeeper来查看或创建topic，新版本使用-bootstrap-server替换老版本的--zookeeper-server

7.2执行创建topic时报错

Error while executing topic command : Timed out waiting for a node assignment. Call: createTopics
image.png
因为访问了错的端口，所以请求失败导致超时。需要使用Kafka端口，本次使用9092

7.3测试消费者时有告警信息，且在生成者端输入信息消费者端未收到信息

WARN [Consumer clientId=console-consumer, groupId=console-consumer-549] Error while fetching metadata with correlation id 2 : {crawl_data=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
image.png
原因是未指定正确的topic，修改成正确的主题名后可正常获取信息

7.4目的端启动DRS服务时报错：未找到动态库文件 path: libdrekafka.so

image.png
【解决方式】
查询libdrekafka.so依赖的动态库
[dmdba@x86_1_59 bin]$ ldd libdrekafka.so
image.png
查找libsasl2.so文件
[root@x86_1_59 ~]# find / -name libsasl2*
查找到后cp一份到drs安装目录bin下
[root@x86_1_59 ~]# cp /usr/lib64/libsasl2.so /home/dmdba/dmdrs5/bin/libsasl2.so.2
[root@x86_1_59 ~]# chown -R dmdba. /home/dmdba/dmdrs5/bin/libsasl2.so.2
[root@x86_1_59 ~]# chmod 755 /home/dmdba/dmdrs5/bin/libsasl2.so.2
再次查询libdrekafka.so依赖的动态库，确认不缺少动态库即可
[dmdba@x86_1_59 bin]$ ldd libdrekafka.so
image.png

上一篇：
达梦大表如何优雅地删数据
评论
后发表回复

作者

文章

阅读量

获赞

扫一扫
联系客服
logo
logo
logo
鄂Copyright © 达梦国产化应用创新实验室
达梦在线服务平台社区用户管理规范

鄂ICP备18017926号-1