1 环境准备
操作系统：centos7.9图形化
zookeeper版本：zookeeper-3.8.3
kafka版本：kafka_2.11-2.0.0.tgz
达梦数据库版本：dm8_20230630_x86_rh6_64.iso
dmhs版本：dmhs_V4.3.18_dm8_kafka_rev136029_rh6_64_20230802.bin
1.1 系统初始化
1.1.1 关闭防火墙和selinux
systemctl stop firewalld && systemctl disable firewalld
sed -i 's/SELINUX=enforcing/SELINUX=disabled/g' /etc/selinux/config && setenforce 0
1.1.2 设置主机名
#192.168.248.100主机设置
hostnamectl set-hostname kafka1
#192.168.248.101主机设置
hostnamectl set-hostname kafka2
#192.168.248.102主机设置
hostnamectl set-hostname kafka3
1.1.3 设置固定IP地址
#192.168.248.100主机设置
vim /etc/sysconfig/network-scripts/ifcfg-ens32
TYPE="Ethernet"
PROXY_METHOD="none"
BROWSER_ONLY="no"
BOOTPROTO="none"
DEFROUTE="yes"
IPV4_FAILURE_FATAL="no"
IPV6INIT="yes"
IPV6_AUTOCONF="yes"
IPV6_DEFROUTE="yes"
IPV6_FAILURE_FATAL="no"
IPV6_ADDR_GEN_MODE="stable-privacy"
NAME="ens32"
DEVICE="ens32"
ONBOOT="yes"
IPADDR=192.168.248.100
DNS1=192.168.248.2
GATEWAY=192.168.248.2
#192.168.248.101主机设置
vim /etc/sysconfig/network-scripts/ifcfg-ens32
TYPE="Ethernet"
PROXY_METHOD="none"
BROWSER_ONLY="no"
BOOTPROTO="none"
DEFROUTE="yes"
IPV4_FAILURE_FATAL="no"
IPV6INIT="yes"
IPV6_AUTOCONF="yes"
IPV6_DEFROUTE="yes"
IPV6_FAILURE_FATAL="no"
IPV6_ADDR_GEN_MODE="stable-privacy"
NAME="ens32"
DEVICE="ens32"
ONBOOT="yes"
IPADDR=192.168.248.101
DNS1=192.168.248.2
GATEWAY=192.168.248.2
#192.168.248.102主机设置
vim /etc/sysconfig/network-scripts/ifcfg-ens32
TYPE="Ethernet"
PROXY_METHOD="none"
BROWSER_ONLY="no"
BOOTPROTO="none"
DEFROUTE="yes"
IPV4_FAILURE_FATAL="no"
IPV6INIT="yes"
IPV6_AUTOCONF="yes"
IPV6_DEFROUTE="yes"
IPV6_FAILURE_FATAL="no"
IPV6_ADDR_GEN_MODE="stable-privacy"
NAME="ens32"
DEVICE="ens32"
ONBOOT="yes"
IPADDR=192.168.248.102
DNS1=192.168.248.2
GATEWAY=192.168.248.2
1.1.4 设置主机名解析
cat <<EOF>>/etc/hosts
192.168.248.100 kafka1
192.168.248.101 kafka2
192.168.248.102 kafka3
EOF
上述都设置完成后重启
selinux需要重启后生效，不过可以临时生效setenforce 0（临时生效命令）
主机名设置需要关闭窗口后重新打开才能生效，因此建议重启服务器使修改配置全部生效
reboot
2 下载安装****zookeeper
2.1 创建相关目录
mkdir -p /soft /opt/bigdata
2.2 zookeeper****下载地址
cd /soft
wget https://dlcdn.apache.org/zookeeper/zookeeper-3.8.3/apache-zookeeper-3.8.3-bin.tar.gz
tar zxvf apache-zookeeper-3.8.3-bin.tar.gz
mv /soft/apache-zookeeper-3.8.3-bin /opt/bigdata/zookeeper-3.8.3
cp -a /opt/bigdata/zookeeper-3.8.3/conf/zoo_sample.cfg /opt/bigdata/zookeeper-3.8.3/conf/zoo.cfg
2.3 修改zoo.cfg配置文件
vim /opt/bigdata/zookeeper-3.8.3/conf/zoo.cfg
#添加serve
server.1=kafka1:28/opt/bigdata88:3888
server.2=kafka2:2888:3888
server.3=kafka3:2888:3888
#添加dataDir
dataDir=/opt/bigdata/data/zookeeper/zkdata
#修改dataLogDir
dataLogDir=/opt/bigdata/data/zookeeper/zkdatalog
2.4 将zookeeper远程拷贝给其他节点
scp -r /opt/bigdata 192.168.248.101:/opt/
scp -r /opt/bigdata 192.168.248.102:/opt/
2.5 配置myid
#192.168.248.100主机设置
echo '1' > /opt/bigdata/data/zookeeper/zkdata/myid
#192.168.248.101主机设置
echo '2' > /opt/bigdata/data/zookeeper/zkdata/myid
#192.168.248.102主机设置
echo '3' > /opt/bigdata/data/zookeeper/zkdata/myid
2.6 配置zookeeper环境变量
vim /etc/profile
#文件最下面添加下列参数
export ZOOKEEPER_HOME=/opt/bigdata/zookeeper-3.8.3
export PATH=$ZOOKEEPER_HOME/bin:$PATH
2.7 生效环境配置
source /etc/profile
2.8 启动zookeeper
zkServer.sh start
2.8.1 查看启动状态
zkServer.sh status
3 下载安装kafka
3.1 下载kafka
cd /soft
wget https://archive.apache.org/dist/kafka/2.0.0/kafka_2.11-2.0.0.tgz
解压安装包到指定路径
tar zxvf kafka_2.11-2.0.0.tgz -C /opt/bigdata/
3.2 修改配置文件
这里重点修改三个参数broker.id标识本机、log.dirs是kafka接收消息存放路径、zookeeper.connect指定连接的zookeeper集群地址
vim /opt/bigdata/kafka_2.11-2.0.0/config/server.properties
#修改下列参数其他默认即可
broker.id=1
log.dirs=/opt/bigdata/kafka_2.11-2.0.0/kafka-logs zookeeper.connect=kafka1:2181,kafka2:2181,kafka3:2181
将kafka包远程拷贝给192.168.248.101/102节点
scp -r /opt/bigdata/kafka_2.11-2.0.0 192.168.248.101:/opt/bigdata/
scp -r /opt/bigdata/kafka_2.11-2.0.0 192.168.248.102:/opt/bigdata/
3.3 修改****server.properties配置文件
#192.168.248.101主机
vim /opt/bigdata/kafka_2.11-2.0.0/config/server.properties
#修改下列参数其余跟192.168.248.100一样
broker.id=2
#192.168.248.102主机
vim /opt/bigdata/kafka_2.11-2.0.0/config/server.properties
#修改下列参数其余跟192.168.248.100一样
broker.id=3
3.4 启动kafka集群
#三个主机都启动
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-server-start.sh -daemon /opt/bigdata/kafka_2.11-2.0.0/config/server.properties &
3.5 查看kafka进程
3.6 kafka基本操作
3.6.1创建topic
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-topics.sh --create --zookeeper kafka1:2181,kafka2:2181,kafka3:2181 --replication-factor 3 --partitions 6 --topic kfk****-****test
3.6.2 列出创建的topic
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-topics.sh --list --zookeeper kafka1:2181,kafka2:2181,kafka3:2181
3.6.3 生成数据
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-console-producer.sh -broker-list kafka1:9092,kafka2:9092,kafka3:9092 --topic kfk-test
3.6.4 消费生产数据
kafka 0.9版本之前用zookeeper
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-console-consumer.sh --zookeeper kafka1:2181,kafka2:2181,kafka3:2181 --from-beginning --topic kfk-test
kafka 0.9版本之后不推荐zookeeper方式,仍然支持，但逐渐会被取消，推荐bootstrap-server方式
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 --from-beginning --topic kfk-test
3.6.5 验证
打开三个节点，其中192.168.248.100打开使用生成数据
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-console-producer.sh -broker-list kafka1:9092,kafka2:9092,kafka3:9092 --topic kfk-test
192.168.248.101/102使用消费生产数据
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 --from-beginning --topic kfk-test
生产数据产生的数据会在消费生产数据哪里显示出来，如下截图
3.6.6 查看指定topic
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-topics.sh --describe --zookeeper kafka1:2181,kafka2:2181,kafka3:2181 --topic kfk-test
可以看到6个分区，每个分区3个副本
partiton： partion id 分区id
leader：当前负责读写的lead broker id ，就是server.properties的broker.id
replicas：当前partition的所有replication broker list
isr：（in-sync replicas）relicas的子集，只包含出于活动状态的broker，离线或挂掉的broker不在此列表
3.6.7 删除指定的topic
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-topics.sh --delete --zookeeper kafka1:2181,kafka2:2181,kafka3:2181 --topic kfk-test
4 安装达梦
（此操作在192.168.248.100操作即可）
4.1 创建相关用户目录
groupadd dinstall
useradd -g dinstall -m -d /home/dmdba/dm -s /bin/bash dmdba
mkdir -p /dbdata/dmdata /dbbak/dmbak /dbarch/dmarch /home/dmdba/dm/dmdbms /mnt/dm /opt/dm
echo "123123" | passwd --stdin dmdba
chown -R dmdba:dinstall /dbdata /dbbak /dbarch /opt/dm /mnt/dm /home/dmdba/
4.2 root用户下修改资源限制
cat <<EOF>> /etc/security/limits.conf
soft core unlimited
hard core unlimited
dmdba soft nofile 65536
dmdba hard nofile 65536
dmdba soft nproc 10240
dmdba hard nproc 10240
dmdba soft core unlimited
dmdba hard core unlimited
EOF
#验证配置是否生效
su - dmdba
ulimit -a
4.3 禁用透明大页
透明大页会随机影响性能，因此建议禁用
4.3.1 临时关闭
echo never > /sys/kernel/mm/transparent_hugepage/enabled
cat /sys/kernel/mm/transparent_hugepage/enabled
修改前
[always] madvise never
修改后
always madvise [never]
4.3.2 永久关闭
需重启服务器后生效
vi /etc/rc.d/rc.local
if test -f /sys/kernel/mm/transparent_hugepage/enabled; then
echo never > /sys/kernel/mm/transparent_hugepage/enabled
fi
if test -f /sys/kernel/mm/transparent_hugepage/defrag; then
echo never > /sys/kernel/mm/transparent_hugepage/defrag
fi
添加执行权限
chmod +x /etc/rc.d/rc.local
4.4 上传安装包
用xshell或者winscp上传安装包到系统/opt/dm下
下列以xshell为例
4.5 安装达梦数据库
（此操作在192.168.248.100操作即可）
4.5.1 挂载安装数据库
mount -o loop /opt/dm/dm8_20230630_x86_rh6_64.iso /mnt/dm
su - dmdba
source .bash_profile
/mnt/dm/DMInstall.bin -i
------------------------------以下为安装过程选项----------------------------------------------------------
Please select the installer's language (E/e:English C/c:Chinese) [E/e]:c #选择c中文安装
解压安装程序..........
欢迎使用达梦数据库安装程序
是否输入Key文件路径? (Y/y:是 N/n:否) [Y/y]:n #选择n后续安装key文件有key可以直接输入绝对路径
是否设置时区? (Y/y:是 N/n:否) [Y/y]:y #选择是
设置时区:
请选择设置时区 21: **** #默认直接回车如果不是21 中国时区填21****
安装类型:
1 典型安装
2 服务器
3 客户端
4 自定义
请选择安装类型的数字序号 [1 典型安装]: #默认直接回车选择典型安装，典型是包含了服务器和客户端安装，自定义是可选择
所需空间: 1869M
请选择安装目录 [/home/dmdba/dm/dmdbms]: #按要求安装路径，有特殊要求填入其他路径
可用空间: 68G
是否确认安装路径(/home/dmdba/dm/dmdbms)? (Y/y:是 N/n:否) [Y/y]: #默认直接回车Y
安装前小结
安装位置: /home/dmdba/dm/dmdbms
所需空间: 1869M
可用空间: 68G
版本信息:
有效日期:
安装类型: 典型安装
是否确认安装? (Y/y:是 N/n:否):
是否确认安装? (Y/y:是 N/n:否):y
2024-01-08 11:03:48
[INFO] 安装达梦数据库...
2024-01-08 11:03:48
[INFO] 安装 基础 模块...
2024-01-08 11:03:53
[INFO] 安装 服务器 模块...
2024-01-08 11:03:54
[INFO] 安装 客户端 模块...
2024-01-08 11:03:57
[INFO] 安装 驱动 模块...
2024-01-08 11:03:59
[INFO] 安装 手册 模块...
2024-01-08 11:03:59
[INFO] 安装 服务 模块...
2024-01-08 11:03:59
[INFO] 移动日志文件。
2024-01-08 11:04:00
[INFO] 安装达梦数据库完成。
请以root系统用户执行命令:
/home/dmdba/dm/dmdbms/script/root/root_installer.sh
安装结束
------------------------------以上为安装过程选项----------------------------------------------------------
再开一个窗口或者切换到root用户下执行上述命令
su - root
/home/dmdba/dm/dmdbms/script/root/root_installer.sh
4.5.2 初始化数据库实例
su - dmdba
/home/dmdba/dm/dmdbms/bin/dminit INI_FILE=/dbdata/dmdata/DAMENG/dm.ini PATH=/dbdata/dmdata extent_size=16 page_size=32 case_sensitive=y log_size=2048
4.5.3 注册数据库服务
/home/dmdba/dm/dmdbms/script/root/dm_service_installer.sh -t dmserver -dm_ini /home/dmdba/dm/dmdbms/bin/dm.ini -p dmserverd
4.5.4 修改初始化数据库实例后创建的配置文件
su - root
mv /usr/lib/systemd/system/DmServicedmserverd.service /usr/lib/systemd/system/dmserverd.service
mv /etc/systemd/system/multi-user.target.wants/DmServicedmserverd.service /etc/systemd/system/multi-user.target.wants/dmserverd.service
mv /home/dmdba/dm/dmdbms/bin/DmServicedmserverd /home/dmdba/dm/dmdbms/bin/dmserverd
vim /usr/lib/systemd/system/dmserverd.service
---------------------------------------------------以下红色是需要修改部分-----------------------------------
[Unit]
Description=DM Instance Service(DmServicedmserverd).
After=network.target remote-fs.target
[Service]
Type=forking
PIDFile=/home/dmdba/dm/dmdbms/bin/pids/dmserverd.pid
ExecStart=/home/dmdba/dm/dmdbms/bin/dmserverd start
ExecStop=/home/dmdba/dm/dmdbms/bin/dmserverd stop
PrivateTmp=true
User=dmdba
TasksMax=infinity
LimitCORE=infinity
LimitNOFILE=100000
LimitNPROC=100000
[Install]
WantedBy=multi-user.target
---------------------------------------------------以上红色是需要修改部分-----------------------------------
4.5.5 启动实例服务
systemctl start dmserverd.service
4.5.6 登录验证
/home/dmdba/dm/dmdbms/bin/disql SYSDBA/SYSDBA
4.5.7 优化命令路径
su - dmdba
cat <<EOF>> /home/dmdba/dm/.bash_profile
export PATH=$PATH:/home/dmdba/dm/dmdbms/bin/
export PATH=$PATH:/home/dmdba/dm/dmdbms/tool/
EOF
生效环境配置
source ~/.bash_profile
验证
5 安装dmhs
mkdir /home/dmdba/dmhs
/opt/dm/dmhs_V4.3.18_dm8_kafka_rev136029_rh6_64_20230802.bin -i
--------------------下列为安装过程所需选择配置-------------------------
Extract install files..........
1.English(English)
2.Simplified Chinese(简体中文)
Select the language to install[2.Simplified Chinese(简体中文)]:2 ##选择2简体中文
/tmp/DMHSInstall/install.log
1.免费试用达梦数据实时同步
2.使用已申请的Key文件
验证许可证文件[1.免费试用达梦数据实时同步]:2 ##选择2
Key文件路径:/opt/dm/dmhs20240925.key #填入授权key路径
Key文件限制信息如下:
有效日期:2024-09-25
用户名称:达梦公司产品试用
服务器颁布类型：试用版
发布类型：企业版
Key类型：企业版
授权用户数：无限制
并发连接数：无限制
1.精简版
2.完整版(web客户端)
3.自定义
安装类型[1.精简版]:1 ##按需求选择，本次选择1
1.实时同步软件服务器
2.远程部署工具
3.实时同步软件配置助手
4.手册
所需磁盘空间:540 MB
安装目录: [/home/dmdba/dmhs]
该路径不为空,是否继续安装?[Y or N]y
安装路径可能存在覆盖安装
1.统一部署
2.现在初始化
是否初始化达梦数据实时同步系统[1.统一部署]:1 #选择1
正在安装
default start ... default finished.
server start ... server finished.
hs_agent start ... hs_agent finished.
hsca start ... hsca finished.
doc start ... doc finished.
postinstall start ... postinstall finished.
正在创建快捷方式
安装成功
远程部署工具配置
远程部署工具名称[HsAgent]: #默认回车
主机Ip(外网)192.168.248.34:192.168.248.34 #填入外网IP地址，如果内网外网两个IP地址的话
远程部署工具管理端口5456: #默认回车
内置数据库轮询间隔3: #默认回车
内置数据库IP[]:192.168.248.34 #有内网IP地址填写内网IP，本次我只设置一个IP地址
内置数据库端口15236: #默认回车
内置数据库用户名[SYSDBA]: #默认回车
内置数据库密码[SYSDBA]: #默认回车
服务脚本环境变量设置
依赖库路径
提示：此配置项供用户配置源或目的数据库依赖库路径和odbc依赖库路径, 多个路径以":"隔开(例:/opt/dmdbms/bin:/usr/local/lib),此配置项会添加到服务脚本的NEED_LIB_PATH的变量值中。
请配置依赖库路径:/home/dmdba/dmhs
远程控制服务
1.自动
2.手动
启动方式:[2.手动]2 #选择2手动
正在创建远程控制服务
达梦数据实时同步V4.0安装完成
更多安装信息，请查看安装日志文件:
/home/dmdba/dmhs/log/install.log
#默认回车结束
--------------------上述为安装过程所需选择配置-------------------------
5.1 开启归档与逻辑日志
disql sysdba/SYSDBA
#开启归档日志
ALTER DATABASE ARCHIVELOG;
ALTER DATABASE ADD ARCHIVELOG 'DEST=/dbarch/dmarch, TYPE=LOCAL, FILE_SIZE=1024, SPACE_LIMIT=51200';
ALTER DATABASE OPEN;
#开启逻辑日志
sp_set_para_value(1,'RLOG_APPEND_LOGIC',1);
select value from v$parameter where name = 'RLOG_APPEND_LOGIC';
#复制libdmoci.so到指定目录
cp -a /home/dmdba/dmhs/db/bin/libdmoci.so /home/dmdba/dmhs/bin
****#****检查依赖
cd /home/dmdba/dmhs/bin
ldd libdmhs_exec.so
5.2 创建dmhs.hs
vim /home/dmdba/dmhs/bin/dmhs.hs
<dmhs>
<base>
<lang>ch</lang>
<mgr_port>5345</mgr_port>
<ckpt_interval>60</ckpt_interval>
<siteid>1</siteid>
<version>2.0</version>
</base>
<cpt>
<db_type>dm8</db_type>
<db_server>192.168.248.100</db_server>
<db_port>5236</db_port>
<db_user>SYSDBA</db_user>
<db_pwd>SYSDBA</db_pwd>
<ddl_mask>0</ddl_mask>
<char_code>PG_UTF8</char_code>
<arch>
<clear_interval>600</clear_interval>
<clear_flag>0</clear_flag>
</arch>
<send>
<ip>192.168.248.100</ip>
<mgr_port>5445</mgr_port>
<data_port>5446</data_port>
<trigger>0</trigger>
<constraint>0</constraint>
<identity>1</identity>
<filter>
<enable>
<item>XGQ.*</item>
</enable>
</filter>
<map>
</map>
</send>
</cpt>
</dmhs>
5.3 在源端数据库建立数据库对象
执行创建促发器语句
执行sql
5.4 创建dmhs的topic
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-topics.sh --create --zookeeper kafka1:2181,kafka2:2181,kafka3:2181 --replication-factor 3 --partitions 6 --topic DMHS
5.5 查看dmhs的topic是否创建
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-topics.sh --describe --zookeeper kafka1:2181,kafka2:2181,kafka3:2181 --topic DMHS
5.6 查看消费者的界面是否有消息推送过来
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-console-consumer.sh --bootstrap-server 192.168.248.100:9092 --topic DMHS --from-beginning
5.7 将kafka目录下的libs目录的文件，全部copy 至dmhs软件bin目录下
cp /opt/bigdata/kafka_2.11-2.0.0/libs/* /home/dmdba/dmhs/bin/
5.8 创建dmhs_kafka.properties
vim /home/dmdba/dmhs/bin/dmhs_kafka.properties
--------------下述dmhs_kafka.properties配置文件内容-------------------------------
DMHS config file path
dmhs.conf.path=/home/dmdba/dmhs/bin/dmhs.hs
这里设置的是kafka集群主机名和端口
bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
kafka topic name
kafka.topic.name=DMHS
#dmhs.sendKey.parse.format=schema:source:tableName
#dmhs.sendKey.parse.format=primary_keys_values
#dmhs.sendTopic.parse.format=schema:source:tableName
#topic.map.conf.path=/dmhs_kafka/bin_0329/tableTopicMap.properties
whether to enable JSON format check
json.format.check=0
How many messages print cost time
print.message.num=1000
How many messages batch to get
dmhs.min.batch.size=100
kafka serializer class
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
kafka partitioner config
partitioner.class=com.dameng.dmhs.dmga.service.impl.OnePartitioner
kafka request acks config
acks=-1
max.request.size=5024000
dmhs.min.batch.size=3000
batch.size=1048576
linger.ms=3
buffer.memory=134217728
retries=3
#enable.idempotence=true
compression.type=none
max.in.flight.requests.per.connection=1
send.buffer.bytes=1048576
metadata.max.age.ms=300000
--------------上述dmhs_kafka.properties配置文件内容-------------------------------
5.9 创建dmhs.hs
vim /home/dmdba/dmhs/bin/dmhs.hs
下列为dmhs.hs配置文件内容这个配置文件适用与dmhs和kafka都在一台机器上------------
<?xml version="1.0" encoding="GB2312" standalone="no"?> <dmhs><base>
<lang>ch</lang>
<mgr_port>5345</mgr_port>
<chk_interval>3</chk_interval>
<ckpt_interval>60</ckpt_interval>
<siteid>1</siteid>
<version>2.0</version>
</base>
<exec>
<recv>
<data_port>5346</data_port>
</recv>
<db_name></db_name>
<exec_thr>1</exec_thr>
<case_sensitive>0</case_sensitive>
<exec_policy>2</exec_policy>
<toggle_case>0</toggle_case>
<commit_policy>1</commit_policy>
<enable_merge>0</enable_merge>
<is_kafka>1</is_kafka>
</exec>
<db_type>dm8</db_type>
<db_server>127.0.0.1</db_server>
<db_user>SYSDBA</db_user>
<db_pwd>SYSDBA</db_pwd>
<db_port>5236</db_port>
<dict_dir>/home/dmdba/dmhs/dmhs_dict</dict_dir>
<char_code>PG_GB18030</char_code>
<ddl_mask></ddl_mask>
<arch>
<clear_interval>600</clear_interval>
<clear_flag>1</clear_flag>
</arch>
<send>
<ip>192.168.248.100</ip>
<mgr_port>5345</mgr_port>
<data_port>5346</data_port>
<net_pack_size>256</net_pack_size>
<net_turns>0</net_turns>
<crc_check>0</crc_check>
<trigger>0</trigger>
<constraint>0</constraint>
<identity>0</identity>
<filter>
<enable>
<item>*.*</item>
</enable>
<disable/>
</filter>
<map>
</map>
</send>
</cpt>
----------------------上述为dmhs.hs配置文件内容---------------------------
5.10 创建json_format.xml
vim /home/dmdba/dmhs/bin/json_format.xml
----------------------下列为json_format.xml配置文件内容---------------------------
char_replace=(","),(,\)
batch_commit=1
gs_flag=1
cur_time_format=(yyyy-mm-ddThh:mi:ss)
commit_time_format=(yyyy-mm-ddThh:mi:ss)
lob_piece=false
clob_format=char
set_null=1
opcmd_len=3
SET_QUOTA=1
need_crlf=false
json_format={"meta":{"time":"#COMMIT_TIME","userid":"#USERID","op":"#OP_TYPE","scn":"#LSN","rowid":"#ROWID","trans":"#TRXID","seq":#MSG_SEQ,"size":#MSG_NUM,"table":"#SCHEMA.#TABLE","idx":"#IDX","posttime":"#TIME"},"data":{#NEW_VALUES},"key":{#OLD_VALUES}}
----------------------上列为json_format.xml配置文件内容---------------------------
参数说明：
char_replace=(“,”),(,):该参数用于定义同步数据如果出现特殊字符‘\’需要通过该参数准换后同步输出为’',否则该特殊字符会丢失。
json_format：该参数用于配置同步后数据输出的格式，可以自定义设置输出格式。
1.5.5准备kafka依赖文件
将kafka软件安装目录下的lib目录下的全部文件打包复制到dmhs目的端的安装目录的bin目录下作为dmhs的依赖文件：
目录结构为：/dmsoft/dmhs/bin/lib/{kafka依赖文件}
5.11 创建DMHS启动脚本
vim /home/dmdba/dmhs/bin/start_dmhs_kafka.sh
#!/bin/sh
#export LANG=zh_CN.GBK
export LD_LIBRARY_PATH=/home/dmdba/dm/dmdbms/bin:/home/dmdba/dmhs/bin/
ulimit -c unlimited
export LANG=zh_CN.UTF-8
java -Djava.ext.dirs="/home/dmdba/dmhs/bin/lib:." com.dameng.dmhs.dmga.service.impl.ExecDMHSKafkaService /home/dmdba/dmhs/bin/dmhs_kafka.properties
5.12 启动dmhs
/home/dmdba/dmhs/bin/start_dmhs_kafka.sh
5.13 启动目的端EXEC模块
/home/dmdba/dmhs/bin/dmhs_console
DMHS>start exec
5.14 在源端配置数据库字典并启动同步
/home/dmdba/dmhs/bin/dmhs_console
DMHS> clear exec lsn
DMHS> COPY 0 "SCH.NAME IS NOT NULL" DICT|LSN
y
DMHS> start cpt
5.15 验证是否同步成功
在192.168.248.100执行创建表数据或者删除更新数据
/home/dmdba/dm/dmdbms/bin/disql sysdba/SYSDBA
create table test (id int,name varchar(20000));
insert into test values(11,'tom');
commit;
在192.168.248.101主机上打开消费者看是否有数据进来
/opt/bigdata/kafka_2.11-2.0.0/bin/kafka-console-consumer.sh --bootstrap-server 192.168.248.100:9092 --topic DMHS --from-beginning
-------------------------------以上部署完成-------------------------------
如需更多资源支持请上达梦技术社区：
达梦数据库 - 新一代大型通用关系型数据库 | 达梦在线服务平台
版权归原作者 小白yu报道 所有， 如有侵权，请联系我们删除。