- 根据上面给出的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.cursors producer = 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.cursors consumer = 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的缘故
为什么我会出现两条重复记录,原因是我生产者程序运行了多次,生产者多运行一次,消费者程序就会多一次查询
还没有评论,来说两句吧...