gbase8a 整合Kafka3.9.1 读取ogg推送文件入库

1.独立安装kafka

2.修改配置文件： 

修改卡开发的3节点IP： 

修改成对应Kafka机器的IP : 每一台机器都需要修改

同理这个配置也是： 

这个配置可参考这个： 

指定Kafka的日志目录： 

3.任意节点执行如下命令获取UUID: bin/kafka-storage.sh random-uuid

每个节点执行启动命令： bin/kafka-storage.sh format -t c-qctGIkT169ucwwGL8a7A -c config/kraft/server.properties

所有节点执行如下命令：bin/kafka-server-start.sh -daemon config/kraft/server.properties 启动

---新建topic : 

---查询进程： 

已知的消息中数据样例： 

在gbase中新建表： BL_MAINT.BL_LOG 执行主键 ：["BL_HISTORY_UUID","SEQ_NUM"]

--启动生产: 
bin/kafka-console-producer.sh --topic BL_MAINT2 --bootstrap-server 172.16.6.230:9092,172.16.6.231:9092,172.16.6.232:9092
 

--启动消费者： 
bin/kafka-console-consumer.sh --topic BL_MAINT --from-beginning --bootstrap-server 172.16.6.230:9092,172.16.6.231:9092,172.16.6.232:9092
 

--- 新建consumer
CREATE KAFKA CONSUMER BL_MAINT_TST TRANSACTION TOPIC
BL_MAINT2 BROKERS '172.16.6.230:9092,172.16.6.231:9092,172.16.6.232:9092';

-- 启动任务
start kafka consumer BL_MAINT_TST;
--停止
stop  kafka consumer BL_MAINT_TST;

----查询状态： 
select * from information_schema.kafka_consumer_status;

注：每一条消息记录按照逗号分隔

可查询结果表 数据已经进入: