已解决
Kafka与MySQL的组合使用
来自网友在路上 156856提问 提问时间:2023-10-20 17:47:07阅读次数: 56
最佳答案 问答题库568位专家为你答疑解惑
- 根据上面给出的student表,编写Python程序完成如下操作:
(1)读取student表的数据内容,将其转为JSON格式,发送给Kafka;
创建Student表的SQL语句如下:
create table student(
sno char(5),
sname char(10),
ssex char(2),
sage int
);
向student表中插入两条记录的SQL语句如下:
insert into student values(‘95001’,’John’,’M’,23);
insert into student values(‘95002’,’Tom’,’M’,23);
启动zookeeper和kafka的服务
编写一个生产者程序mysql_producer.py:
from kafka import KafkaProducer
import json
import pymysql.cursorsproducer = KafkaProducer(bootstrap_servers='localhost:9092',value_serializer=lambda v:json.dumps(v).encode('utf-8'))connect=pymysql.Connect(host='localhost',port=3306,user='root',passwd='123456',db='zhangna',charset='utf8'
)
cursor=connect.cursor()
sql="select sno,sname,ssex,sage from student;"
cursor.execute(sql)
data=cursor.fetchall()
connect.commit()for message in data:zn={}zn['sno']=message[0]zn['sname']=message[1]zn['sex']=message[2]zn['age']=message[3]producer.send('mysql_topic',zn)connect.close()
producer.close()
(2)再从Kafka中获取到JSON格式数据,打印出来;
编写一个消费者程序mysql_consumer.py:
from kafka import KafkaConsumer
import json
import pymysql.cursorsconsumer = KafkaConsumer('mysql_topic',bootstrap_servers=['localhost:9092'],group_id=None,auto_offset_reset='earliest')
for msg in consumer:msg1=str(msg.value,encoding="utf-8")data=json.loads(msg1)print(data)
终于出来了,出错的原因是encoding,我写成了encodings的缘故
为什么我会出现两条重复记录,原因是我生产者程序运行了多次,生产者多运行一次,消费者程序就会多一次查询
查看全文
99%的人还看了
相似问题
- rabbitMQ的扇出模式(fanout发布订阅)的生产者与消费者使用案例
- Java编写简易rabbitmq生产者与消费者
- SpringBoot Kafka生产者 多kafka配置
- linux入门---消费者生产者模型模拟实现
- java经典例题:生产者/消费者问题
- RocketMQ生产者消息发送出去了,消费者一直接收不到怎么办?(Rocket MQ订阅关系一致性)
- <多线程章节九>生产者消费者模型,阻塞队列的使用和模拟实现
- Spring Kafka生产者实现
- Sentinel学习(2)——sentinel的使用,引入依赖和配置 对消费者进行流控 对生产者进行熔断降级
- 【Linux】生产者和消费者模型
猜你感兴趣
版权申明
本文"Kafka与MySQL的组合使用":http://eshow365.cn/6-20328-0.html 内容来自互联网,请自行判断内容的正确性。如有侵权请联系我们,立即删除!