Hubble数据库数据同步开启

Hubble数据库数据同步开启

root@ty-bj03-test04:35432/defaultdb> SET CLUSTER SETTING kv.rangefeed.enabled = true;
SET CLUSTER SETTING

Time: 256ms total (execution 256ms / network 0ms)

root@ty-bj03-test04:35432/defaultdb>

创建测试表

root@ty-bj03-test04:35432/defaultdb> create table cdc_demo(id int PRIMARY KEY ,name string,tel string,sex string ,index(name));
CREATE TABLE

Time: 271ms total (execution 270ms / network 0ms)

root@ty-bj03-test04:35432/defaultdb> 

创建对应这个表的改变监听，数据同步到 kafka

• updated 默认
• resolved 对应表没有修改数据，也会实时往 kafka topic 写入空数据
• full_table_name 以防不同库有相同表，对应 kafka 会创建带有模式的 topic
• format 同步到 kafka 的格式 json / avro / csv；默认 json

root@ty-bj03-test04:35432/defaultdb> create changefeed for table cdc_demo into 'kafka://192.168.100.134:9092' with updated;
        job_id
----------------------
  813772285347528708
(1 row)

Time: 541ms total (execution 540ms / network 1ms)

root@ty-bj03-test04:35432/defaultdb> 

插入表数据

insert into cdc_demo values(1,'杨x','1111','男'),(2,'刘x','2222','女');
root@ty-bj03-test04:35432/defaultdb> insert into cdc_demo values(1,'杨x','1111','男'),(2,'刘x','2222','女');
INSERT 2

Time: 8ms total (execution 8ms / network 1ms)

root@ty-bj03-test04:35432/defaultdb>

查看对应表的 kafka 的 topic

[hubble@hadoop1 kafka_2.12-3.1.1]$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic cdc_demo 
{"after": {"id": 1, "name": "杨x", "sex": "男", "tel": "1111"}, "updated": "1668414124878734547.0000000000"}
{"after": {"id": 2, "name": "刘x", "sex": "女", "tel": "2222"}, "updated": "1668414124878734547.0000000000"}

Hubble 数据库取消 cdc 任务

root@ty-bj03-test04:35432/defaultdb> cancel job 813776060113190916;
CANCEL JOBS 1

Time: 61ms total (execution 60ms / network 1ms)

root@ty-bj03-test04:35432/defaultdb>

转载 Hubble数据库数据同步开启