RTSync安装部署

本次演示的是利用rtsync将mysql数据实时同步至GBase8a MPP。

1. 安装包列表

jdk-8u211-linux-x64.tar.gz
zookeeper-3.4.9.tar.gz                # 可不安装，kafka自带
kafka_2.11-0.10.2.1.tgz
GBase_RTSync_8.6.24.0_1.tar.gz
mysql-5.7.39-el7-x86_64.tar.gz 
GBase8a_MPP_Cluster-NoLicense-8.6.2_build43-R34-2-redhat7-x86_64.tar.bz2

2. 安装各组件并启动

各组件安装过程（略）。

cd /usr/local/hadoop/kafka/bin
# 启动zk
cd /usr/local/hadoop/kafka && ./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties
# 启动kafka
cd /usr/local/hadoop/kafka && ./bin/kafka-server-start.sh -daemon ./config/server.properties

# 创建topic
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic topic01
# 查看topic列表
./kafka-topics.sh --list --zookeeper localhost:2181
# 详细信息
./kafka-topics.sh --describe --zookeeper localhost:2181

# 启动mysql
/usr/local/mysql/support-files/mysql.server start
# 启动RTSync
sh /opt/RTSync/RTSyncManagerServer.sh start

3. rtsync配置文件

3.1 配置consumer任务文件

vi RTSync/conf/config_task.xml

<?xml version="1.0" encoding="UTF-8" ?>
<servers>
   <!-- # isHighAvailable="true" dataRecoveryMode="zookeeper" 开启高可用 -->
   <server id="server2" syncMode="increment" dataFormatType="PUREDATA" mqType="kafka" queueName="mysqlto8a" dataRecoveryMode="file" isHighAvailable='false'>
   <!-- 管理端IP -->
       <manager ip="192.168.10.101"
                    port="9432"
                    heartbeatPort="9000"
                    httpPort="8085"
                    heartbeatTimeOut="1800"
                    isTableHotPatch="true"
                    hotPatchTimeOut="30"
                    useErrorDataRecovery="false"/>
       <!-- #读端部署服务器IP -->
       <source ip="192.168.10.101"
                    path="/opt/RTSync"
                    queueSize="5"
                    readParseAdapter="adapter"
                    user="root"
                    password="root"
                    openMonitor="true"
                    monitorInterval="300"
                    dataFormatParallel="10"
                    rpcPort="9191"
                    sshPort="22"
                    isDropTmpTable="true"
                    dbObjToUpperCase="true"
                    isConvertSingleQuote="false"
                    isAllowInsertPkNull="true"
                    isisEmptyStrPkEqualsNull="true"/>
       <mappings>
           <source-target id="mysqlto8a">
               <db>
                   <sourcedb allowPrimaryKeyNull="true"
                              catalog="testdb"
                              charset="UTF8"
                              columnTypeFormat="time=hh:mm:ss.SSSSSSS;date=yyyy-MM-dd;datetime=yyyy-MM-dd HH:mm:ss;smalldatetime=yyyy-MM-dd HH:mm:ss.SSSSSS;datetime2=yyyy-MM-dd HH:mm:ss.SSSSSS"
                              delayTimeThreshold="0"
                              driver="com.mysql.jdbc.Driver"
                              dyntal="true"
                              fetchSize="100"
                              isGetDBspace="true"
                              isMineWholeTransaction="fasle"
                              isOverDelayTimeThresholdExit="false"
                              isParallelForMysql="true"
                              isUseMinerCache="true"
                              longTxCheckIntervalTime="10"
                              longTxMaxWaitTime="60"
                              maxRecordsPerRead="200"
                              maxSizeOfPerRecord="1024"
                              mysqlSlaveId="00"
                              operationType="dml"
                              packetMaxRecord="2"
                              parallel="4"
                              password="root"
                              startLSN="0"
                              timeOut="2"
                              timestampWithFraction="true"
                              transMaxCount="10000"
                              type="MYSQL"
                              url="jdbc:mysql://localhost:3306/testdb?useSSL=false"
                              useAddFile="false"
                              user="root" />
                   <targetdb
                       charset="UTF8"
                       type="GCLUSTER"
                       commitSize="10000"
                       queueSize="200000"
                       timeOut="2"
                       user="gbase"
                       password="gbase20110531"
                       driver="com.gbase.jdbc.Driver"
                       catalog="testdb"
                       url="jdbc:gbase://192.168.10.102:5258/gbase?useOldAliasMetadataBehavior=true&rewriteBatchedStatements=true&connectTimeout=0&socketTimeout=0" />
                   <tables isInclude="true">
                       <table  deleteMode="NORMAL"  sourceTableName="testdb.t1" sourcePkColName="id" targetTableName="testdb.t1" targetPkColName="id">
                           <column-mapping sourceColName="id" targetColName="id"/>
                           <column-mapping sourceColName="name" targetColName="name"/>
                       </table>
                   </tables>
               </db>
           </source-target>
       </mappings>
   </server>
</servers>

3.2 配置kafka topic

vi RTSync/conf/config_kafka_mysqlto8a.properties

kafka.producer.paramers=request.timeout.ms=600000;metadata.fetch.timeout.ms=60000;linger.ms=10;
kafka.send.issuccess=true
kafka.consumer.paramers=session.timeout.ms=10000;request.timeout.ms=60000;enable.auto.commit=false;max.poll.records=100;max.partition.fetch.bytes=909200000
zookeeper.session.timeout.ms=8000
kafka.resend.max.retries=3
topic.replication.num=1
send.data.max.size=30485760
kafka.batch.commit.time=300
fetch.message.max.bytes=104857600
group.id=mysqlto8a
zookeeper.sync.time.ms=4000
bootstrap.servers=192.168.10.101:9092
auto.commit.offset.enable=false
kafka.acks=all
kafka.batch.commit.count=1000
topic.enable.auto.create=true
topic.name=football
auto.commit.interval.ms=5000
zookeeper.connect=192.168.10.101:2181

2.1.4 GBase 8a参数配置

2.1.4.1 gcluster层

vi /opt/gcluster/config/gbase_8a_gcluster.cnf

gcluster_kafka_consumer_enable=1 
_gbase_transaction_disable=1
_gcluster_insert_cache_buffer_flag=1
gcluster_assign_kafka_topic_period=20
gcluster_kafka_max_message_size=10000        # 内存较小调小否则会导致数据库宕机
gcluster_kafka_batch_commit_dml_count=1000
gcluster_kafka_local_queue_size=2100
gcluster_kafka_user_allowed_max_latency=20000
gcluster_kafka_consume_batch=100
gcluster_kafka_parallel_commit=5
gcluster_kafka_primarykey_can_be_null=1
gcluster_kafka_data_buf_size=200
gcluster_kafka_message_format_type=PUREDATA

2.1.4.2 gnode层

vi /opt/gnode/config/gbase_8a_gbase.cnf

_gbase_transaction_disable=1            # 注意一定不要用0
gbase_tx_log_mode=ONLY_SPECIFY_USE        # 注意一定不要用USE,STANDARD_TRANS，V95删除此参数
_gbase_kafka_transaction_mode=1         # 95版本此参数
gbase_buffer_insert=256M                # 内存小则参数设置小一些
gbase_tx_log_flush_time=5

2.1.4.3 重启服务

service gcware restart

2.1.5 GBase8a创建消费任务

-- 创建任务: topic01与kafka中创建的topic必须一致
Create kafka consumer kc01 transaction topic topic01 brokers '192.168.10.101:9092';
-- 启动/停止任务
start kafka consumer kc01;
stop kafka consumber kc01;
drop kafka consumber kc01;
show kafka consumber kc01;
-- 系统表
select * from information_schema.kafka_consumer_status; 
select * from  gclusterdb.kafka_consumers;

2.1.6 测试验证

# mysql端
mysql> truncate table testdb.t1;
Query OK, 0 rows affected (0.01 sec)

mysql> select * from testdb.t1;
Empty set (0.00 sec)

mysql> insert into testdb.t1 values(round(rand()*1000),'ee');
Query OK, 1 row affected (0.01 sec)

mysql> select * from testdb.t1;
+-----+------+
| id  | name |
+-----+------+
| 256 | ee   |
+-----+------+
1 row in set (0.00 sec)

# gbase端
gbase> select TOPIC,STATUS,MIN_OFFSET,MAX_OFFSET,CUR_OFFSET,PROCESS_OFFSET,COMMIT_OFFSET from information_schema.kafka_consumer_status;
+----------+--------+------------+------------+------------+----------------+---------------+
| TOPIC    | STATUS | MIN_OFFSET | MAX_OFFSET | CUR_OFFSET | PROCESS_OFFSET | COMMIT_OFFSET |
+----------+--------+------------+------------+------------+----------------+---------------+
| football | START  |          0 |         43 |         43 |             42 |            42 |
+----------+--------+------------+------------+------------+----------------+---------------+
1 row in set (Elapsed: 00:00:00.02)

gbase> select * from testdb.t1;
+------+------+
| id   | name |
+------+------+
|  256 | ee   |
+------+------+
1 row in set (Elapsed: 00:00:00.01)