使用DMHS同步DM8到kafka

一、环境介绍
源端：localhost
目的端：149.0.161.73
二、安装
达梦8数据库安装：
ODBC驱动安装：
DMHS使用odbc驱动获取数据，所以源端需要安装odbc驱动

编译安装三部曲：
使用root用户：1、解压后目录执行 ./configure
2、make
3、make install
连接达梦数据库测试
两个配置文件 :在 /usr/local/etc目录下

dmdba@161-73:/usr/local/etc$ cat odbc.ini 
[DM8]
Description = DM ODBC DSN
Driver = DM8 ODBC DRIVER
SERVER = localhost
UID = SYSDBA
PWD = SYSDBA
TCP_PORT = 5236
dmdba@161-73:/usr/local/etc$ cat odbcinst.ini 
[DM8 ODBC DRIVER]
Description = ODBC DRIVER FOR DM8
Driver = /home/dmdba/dmdbms/drivers/odbc/libdodbc.so

dmdba用户环境变量配置，需要访问到odbc共享库，

dmdba@161-73:~$ cat .bash_profile  
export DMHS_HOME=/home/dmdba/dmhs
export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/home/dmdba/dmhs/bin:/home/dmdba/dmoci:/usr/local/lib

访问测试

dmdba@161-73:~$ isql dm8
open fail! path:./libcrypto.so
+---------------------------------------+
| Connected!                            |
|                                       |
| sql-statement                         |
| help [tablename]                      |
| quit                                  |
|                                       |
+---------------------------------------+
SQL> 

连接成功
三、 hs源端安装
hs源端配置文件：

root@161-73:/home/dmdba/dmhs/bin# cat dmhs.hs 



        ch
        5345
        60
        1
2.0


        DM8
        localhost
        SYSDBA
        SYSDBA
        5236
        10
        1
        OP:OBJ:REC
        PG_UTF8
        
                60
                0
        
        
                149.0.161.73
                5345
                5346
                0
                1
                1
                1
                
                        
                                ajgl.*
                        
                        
                
                                
        

四、 kafka安装（安装在执行端）
解压到/usr/local/kafka下：

 tar -xvf kafka_2.13-2.8.1.tgz 

启动zookeeper:kafka启动依赖于zookeeper，单机的使用kafka自带的zookeeper即可，

/usr/local/kafka/config# /usr/local/kafka/bin/zookeeper-server-start.sh  zookeeper.properties

启动kafka

 /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties

创建主题：

/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2182 --replication-factor 1 --partitions 1 --topic dmtest

开启消费者以便查看效果

/usr/local/kafka/bin/kafka-console-consumer.sh  --bootstrap-server 127.0.0.1:9092 --topic dmtest --from-beginning

五、hs_kafka执行端安装
hs_kafka执行端配置文件

kafka需要三个配置文件,分别如下

/opt/dmhs_kafka/bin$ cat dmhs.hs 


        
                ch
                6345
                3
                60
                20
                2.0
        
        
                
                        6346
                
                
                1
                0
                2
                0
                1
                0
                1
                1
                PG_UTF8
        
 
:/opt/dmhs_kafka/bin$ cat dmhs_kafka.properties 
# DMHS config file path
dmhs.conf.path=/opt/dmhs_kafka/bin/dmhs.hs
# kafka broker list,such as ip1:port1,ip2:port2,...
#bootstrap.servers=223.254.16.75:9092,223.254.16.76:9092
bootstrap.servers=127.0.0.1:9092
# kafka topic name
kafka.topic.name=dmtest
# whether to enable JSON format check
json.format.check=1
# How many messages print cost time
print.message.num=1000
# How many messages batch to get
dmhs.min.batch.size=20
# 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
#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
 
启动脚本
 
/opt/dmhs_kafka/bin$ cat start_dmhs_kafka.sh 
#!/bin/sh
export LANG=en_US.UTF-8
export LD_LIBRARY_PATH=/home/dmdba/dmdbms/bin:/home/dmdba/dmoci:/opt/dmhs_kafka/bin 
java -Djava.ext.dirs="/usr/local/kafka/libs:." com.dameng.dmhs.dmga.service.impl.ExecDMHSKafkaService  /opt/dmhs_kafka/bin/dmhs_kafka.properties
 
其中"/usr/local/kafka/libs:.为kafka安装路径，需要kafka自己的jar包
配置文件：
 
六、装载测试
 启动源端
 
 启动目的端
 
 装载测试
 hs消费日志
 
 kafka消费信息