活动专区

活动公告

上车！GBase 8a开发者联盟发车了 技术福利、人脉资源全都有！

认证培训专区

认证培训资料

GBase 8s GDCA 认证培训课前准备

GBase 8a

其他

物化视图报错：references invalid table(s) or column(s) or function(s) or definer/invoker of view lack rights to use them

01-GBase 8s 事务型数据库 安装与卸载

GBase 8s

适配迁移

[GBase 8s 教程]GBase 8s 常用函数、表达式

本文针对南大通用GBase 8c数据库，围绕逻辑解码增量数据推送至Kafka场景，详细介绍配置流程、操作步骤及注意事项，指导用户快速完成增量数据实时同步到Kafka的落地部署。

## 概述

GBase 8c提供逻辑解码能力，通过解析WAL预写日志生成逻辑日志，目标数据库可基于逻辑日志实现实时数据复制。逻辑复制方式对目标数据库部署形态约束较低，支持异构数据库、同库不同部署形态之间的数据同步；同步过程中目标数据库可正常读写，整体同步时延较低。
依托逻辑复制机制，结合GBase 8c自研的逻辑转Kafka能力，可按数据节点并发从WAL日志中实时解析增量数据，并推送至Kafka集群。

## 配置步骤

当前测试是将GBase8c集群中testa库下的public.t_user的数据变更操作推送到kafka，环境信息如下：

| 配置项目       | 配置地址                                           |
| ---------------- | ---------------------------------------------------- |
| 推送的数据名称 | testa                                              |
| 推送表         | public.t_user                                      |
| Kafka 地址     | 172.16.5.51:9092,172.16.5.53:9092,172.16.5.54:9092 |
| Topic 名称     | gbase_user_topic                                   |

### kafka配置

配置时只需提供Kafka地址、Topic名称，若开启SSL安全认证，同步提供kafka认证信息即可。

### 数据库配置

1. 修改数据库配置参数，开启参数后，数据库即可支持将数据变更推送至Kafka。

```
--开启l2k同步开关
gs_guc reload -Z datanode -N all -I all -c "enable_l2k=true"

--指定wal日志存储的级别为logical，在replica级别（支持wal归档和复制）的基础上，还支持逻辑解码信息
gs_guc reload -Z datanode -N all -l all -c "wal_level=logical"

--开启默认用户的replication连接认证
gs_guc reload -Z datanode -N all -l all -h "local  replication gbase trust"
```

2. 新建配置表
    配置表用于记录需要将数据变更推送至Kafka的表清单，必须以复制表形式创建。如：

```
create table gbase_to_kafka_cnf(
	tablename varchar(100),
	topic varchar(100)
) 
DISTRIBUTE BY REPLICATION;

字段解释：
Tablename：需要推送表的名称，字段名称不能修改，字段长度可以调整；
Topic：推送到kafka的topic名称，字段名称不能修改，字段长度可以调整；
```

3. 配置表写入数据
   将需要推送的表清单写入配置表，表名必须按照schema.table_name格式填写。

```
INSERT INTO gbase_to_kafka_cnf values('public.t_user,'gbase_user_topic');
```

4. 开启数据同步

```
gs_guc reload -Z datanode -N all -I all -c "l2k_conf='{\"config\":[{\"dbname\":\"testa\",\"kafka_broker\":\"172.16.5.51:9092,172.16.5.53:9092,172.16.5.54:9092\",\"kafka_topic_table\":\"gbase_to_kafka_cnf\"}],\"type\":\"wal2json\",\"verbose\":\"off\"}'"
```

其他配置项如：

| 配置信息          | 描述                                                              | 举例              |
| ------------------- | ------------------------------------------------------------------- | ------------------- |
| dbname            | 数据库名称。                                                      | test              |
| kafka_broker      | kafka 的 broker 配置，用","分隔。                                 | 192.168.1.10:9092 |
| kafka_topic       | kafka 的 topic 配置，与 kafka_topic_table 互斥。                  | test1             |
| kafka_topic_table | 记录表名和kafka topic 映射关系的数据库表名，与 kafka_topic 互斥。 | testft1           |
| schema2db         | 为schema 与 db 的映射关系，用","分隔；schema 和 db 间用":"分隔。  | test1:public      |
| type              | 转换类型。不填写为默认的wal2json。                                | wal2json          |
| verbose           | 详细日志。默认关闭。                                              | on                |
| pause             | 同步暂停开关。默认关闭。                                          | on                |

其中kafka_topic_table数据表格式要求：

* kafka_topic_table数据表需为复制表，包含tablename和topic列。
* tablename需填写转换后的表名。例如：若配置schema2db转换规则为test1::public，则表test1.k1需填写为public.k1。
  
### 测试推送
 
1. 在数据库中对public.t_user表进行数据DML操作。

```
update t_user set age=28,city='深圳',update_time='2026-05-11 15:38:43.150452' where id=9;
insert into t_user values(12,'王五',28,'北京','2026-05-11 16:55:09.453177');
delete from t_user where id=6;
```

2. 在kafka中读取内容，内容为JSON文档格式，其他应用程序可以通过消费kafka内容，读取到指定表的数据变更。

```
{"change":[{"kind":"update","schema":"public","table":"t_user","columnnames":["id","name","age","city","update_time"],"columntypes":["integer","character varying(50)","integer","character varying(50)","timestamp without time zone"],"columnvalues":[9,"赵六",28,"深圳","2026-05-11 15:38:43.150452"],"oldkeys":{"keynames":["id"],"keytypes":["integer"],"keyvalues":[9]}}]}      {"change":[{"kind":"insert","schema":"public","table":"t_user","columnnames":["id","name","age","city","update_time"],"columntypes":["integer","character varying(50)","integer","character varying(50)","timestamp without time zone"],"columnvalues":[12,"王五",28,"北京","2026-05-11 16:55:09.453177"]}]}     {"change":[{"kind":"delete","schema":"public","table":"t_user","oldkeys":{"keynames":["id","name","age","city","update_time"],"keytypes":["integer","character varying(50)","integer","character varying(50)","timestamp without time zone"],"keyvalues":[6,"张三",25,"南京","2026-05-11 15:38:43.150452"]}}]}
```

## 注意事项

* 默认情况下，推送到Kafka的UPDATE数据仅包含主键列及更新后的数据，不包含其他字段与修改前数据。若需要将修改前的值 + 全部字段一同推送至kafka，需将推送表的replica属性设置为full，命令如下：

```
alter table table_name replica identify full;
```

* 需要注意DN节点的max_replication_slots和max_replication_slots参数，需满足：max_wal_senders>=物理复制槽数量+max_replication_slots。



南大通用GBase 8c逻辑解码概述

评论

热门帖子