debezium source端同步海量数据库数据vastbase到kafka
Original 韦家富 心流时刻 2024年01月30日 15:17 北京本文用于debezium source端同步海量vastbase数据库的数据到kafka，sink端同理。
1、基础环境安装
1、kafka
2、zookeeper
3、海量数据库vastbase
自行安装并配置基础环境。
海量数据库的安装文档：
https://docs.vastdata.com.cn/zh/docs/VastbaseG100Ver2.2.15/doc/%E5%AE%89%E8%A3%85%E5%8D%87%E7%BA%A7%E6%8C%87%E5%8D%97/%E5%8D%95%E6%9C%BA%E5%AE%89%E8%A3%85/%E5%AE%9E%E4%BE%8B%E5%8C%96%E6%95%B0%E6%8D%AE%E5%BA%93%E5%AE%89%E8%A3%85.html
2、修改海量数据库的配置
1、修改postgresql.confvim /home/vastbase/data/vastbase/postgresql.conf
wal_level = hot_standby 改成 wal_level=logical
2、修改pg_hba.conf
vim /home/vastbase/data/vastbase/pg_hba.conf （是控制主机访问的）
最后一行加上： host replication all 0.0.0.0/0 md5
再启动：vb_ctl start
3、控制台执行：
SELECT*FROM pg_create_logical_replication_slot('vastbase_slot', 'pgoutput');
CREATE PUBLICATION dbz_publication FORALLTABLES;
查看系统slot视图：select * from pg_replication_slots ;
其中：vastbase_slot字段 为自定义，后续配置需要
3、下载并编译debezium
下载地址：
https://gitee.com/opengauss/debezium
注：
1、需要JDK11、Apache Maven 3.6.3 及以上
2、编译前，会有各种报错，将test包删除，其它无需处理。
需要编译的目录：
debezium-core
debezium-connector-opengauss
debezium-api
SQL查看pgsql的版本，如果是低于9.6的，则需要修改源码。
SELECT version();
***修改源码处：
debezium-connector-opengauss：
将所有confirmed_flush_lsn 修改为confirmed_flush
编译命令：
mvn clean package -Dmaven.test.skip=true
或者idea 的右边maven编译
最后所需的Jar包及依赖文件（一个都不能少），有些Jar包在编译后就更新repository中了，可自行复制。
4、配置并启动Kafka Connect
1、CD到kafka目录编辑相关配置文件
cd /opt/service/kafka_2.13-3.6.1/config
vi connect-distributed.properties
plugin.path=/opt/service/kafka_2.13-3.6.1/plugins
2、将Jar包目录拷贝到指定的plugins目录下
3、启动Kafka 及 Kafka Connect ：
启动kafka
./kafka-server-start.sh -daemon ../config/server.properties
启动Kafka Connect
./connect-distributed.sh ../config/connect-distributed.properties
Kafka Connect 出现一些错可以不用管
5、注册Kafka Connect
用curl 或者postman等工具，post请求kafka地址注册：
post http://x:x:x:x:8083/connectors
{
"name": "connect-opengauss-source",
"config": {
"connector.class": "io.debezium.connector.opengauss.OpengaussConnector",
"database.hostname": "x.x.x.x",
"database.port": "5432",
"database.user": "vbadmin",
"database.password": "xxxx",
"database.dbname": "vastbase",
"topic.prefix": "vastbase-",
"table.include.list": "public.*", //public下的所有的表
"database.server.id": "1",
"database.server.name": "opengauss",
"tasks.max": "1",
"slot.name": "vastbase_slot", // 定义的slot
"plugin.name": "pgoutput",
"transforms": "route",
"transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.route.regex": "^opengauss(.*)",
"transforms.route.replacement": "opengauss_server_topic",
"decimal.handling.mode": "string",
"include.unknown.datatypes": "true",
"slot.drop.on.stop": "true",
"snapshot.mode": "initial", //此处必须配置，快照模式
"commit.process.while.running": "true",
"source.process.file.path": "/home/vastbase/data/connectors/process/",
"commit.time.interval": "1",
"create.count.info.path": "/home/vastbase/data/connectors/countInfo/",
"process.file.count.limit": "10",
"process.file.time.limit": "168",
"append.write": "false",
"file.size.limit": "10",
"export.csv.path": "/home/vastbase/data/connectors/csv/",
"export.csv.path.size ": " 2G",
"signal.kafka.bootstrap.servers":"x.x.x.x:9092", // 此处为kafka地址
"errors.tolerance":"all" // 此处必须配置，静默忽略无效的消息
}
}
具体字段描述请参考:
debezium-connector-opengauss 下的opengauss-source.properties
请求成功后会返回请求数据：
如想查看，直接用get命令：
如想删除该connectors，直接用delete命令，后续是该connectors
6、成功示例
Kibana Connect 控制台
增删改数据后的kafka控制台：
可用命令：
kafka-console-consumer.sh --bootstrap-server <kafka_broker_address> --topic <topic_name> --from-beginning --property print.key=false --property print.value=true
7、注意事项（容易出问题的地方）
1、Jar包缺失，或者版本过高
2、connectors配置文件配置不正确3、海量数据库配置不正确
4、源码未适配老版的pg slot视图的字段
有其它问题欢迎与我留言沟通。