项目背景

（1）采用数据同步软件将各系统的数据库下的数据实时同步到openGauss数据库中；

（2）建立实时数仓；

（3）可以在实时数仓自行查询、分析、统计数据及报表；

（4）同时横向集成公共服务区和核心业务区生产库数据、集成其他委办局数据。

（5）纵向集成市级的生产库数据和向省级交换库同步数据；

（6）协同国家数据中台的数据归集工作；

（7）并通过高灵活、高性能的优势，也可进行数据的分析、挖掘以及二次应用。

同步架构图

（1）架构描述：源端华为云rds mysql 5.7.33 ,目标端是openGauss5.0。

（2）源端mysql是DDM分库分表架构，

（3）需要把一张表切分成多张表后同步到openGauss 一张表里。

（4）同步架构属于多对一。

下图deb:Debezium

环境介绍

工具准备

kafka

Debezium MySQL Connector

JDBC sink Connector

（1）源端mysql参数配置

log_bin=on
binlog_format=row
binglog_row_image=full

（2）目标端openGauss参数配置

wal_level=logical
上传wal2json
cd $GAUSSHOME/lib/postgresql
ls wal2json.so
wal2json.so

（3）替换jar包

debezium-connector-postgres-1.4.0-SNAPSHOT.jar
postgres.jar

源端和目标端测试数据准备

源端数据准备如下：

nmsimis_0.ac09_0

nmsimis_1.ac09_0

nmsimis_2.ac09_0

源端用华为DDM分库分表工具已经把ac09表切分成3个库中，如下:

mysql> use nmsimis_0;
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A

Database changed
mysql> show tables;
+---------------------+
| Tables_in_nmsimis_0 |
+---------------------+
| ac09_0                |
+---------------------+
1 rows in set (0.00 sec)

mysql> use nmsimis_1;
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A

Database changed
mysql> show tables;
+---------------------+
| Tables_in_nmsimis_1 |
+---------------------+
| ac09_0                |
+---------------------+
1 rows in set (0.00 sec)
mysql> use nmsimis_2;
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A

Database changed
mysql> show tables;
+---------------------+
| Tables_in_nmsimis_2 |
+---------------------+
| ac09_0                |
+---------------------+
1 rows in set (0.00 sec)

mysql> show create table ac09_0;
+--------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Table  | Create Table                                                                                                                                                                                                                 |
+--------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| ac09_0 | CREATE TABLE `ac09` (
 `id` varchar(30) NOT NULL,
`xm` varchar(50) DEFAULT NULL,
`dz` varchar(50) DEFAULT NULL,
`csrq` datetime DEFAULT NULL,
PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 |
+--------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row in set (0.00 sec)

目标端：openGauss

数据库名：dbserver1 模式：nmsimis

项目实施

1. 启动kafka和deb

/dbms/kafka/bin/zookeeper-server-start.sh -daemon /dbms/kafka/config/zookeeper.properties 
/dbms/kafka/bin/kafka-server-start.sh -daemon /dbms/kafka/config/server.properties 
/dbms/kafka/bin/connect-distributed.sh -daemon /dbms/kafka/config/connect-distributed.properties

2. 查看deb支持的插件

[root@k8s-node1 ~]#  curl -s 192.168.10.3:8083/connector-plugins|jq
[
{
"class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"type": "sink",
"version": "5.3.2"
},
{
"class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"type": "source",
"version": "5.3.2"
},
{
"class": "io.debezium.connector.mysql.MySqlConnector",
"type": "source",
"version": "1.5.4.Final"
},
{
"class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
"type": "sink",
"version": "2.8.2"
},
{
"class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
"type": "source",
"version": "2.8.2"
},
{
"class": "org.apache.kafka.connect.mirror.MirrorCheckpointConnector",
"type": "source",
"version": "1"
},
{
"class": "org.apache.kafka.connect.mirror.MirrorHeartbeatConnector",
"type": "source",
"version": "1"
},
{
"class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
"type": "source",
"version": "1"
}
]

3. 创建连接器

（1）配置源端mysql连接器，如下图

（2）配置目标端

curl -s http://192.168.10.3:8083/connectors/openGauss-ac09-connector -X GET|jq

4. 查看连接器状态

[root@k8s-node1 ~]# curl -s http://192.168.10.3:8083/connectors/ -X GET|jq
[
"mysql-ac09-connector",
"openGauss-ac09-connector"
]

curl -s http://192.168.10.3:8083/connectors/openGauss-ac09-connector/status -X GET|jq

5. 验证表初始化

通过登录openGauss查看表结构已经同步过来，如下图

6. 验证DML实时同步

源端:插入数据

查看目标端：

总结

构建实时数据仓库的关键之一就是设计一个可靠而强大的数据库架构。这就像在峡谷中建造一座坚固的桥梁，让你的数据可以安全地流动。openGauss可以轻松应对海量数据的处理和存储。就像一位巧妙的工程师，openGauss在背后默默助力，为你的数据仓库提供强大的基础。它的灵活性和扩展性让你能够根据业务需求精确地设计数据模型，确保数据的准确性和一致性。