沈阳网站备案,如何建设国外的网站,导航网站开发工具,wordpress网站搬根据上面给出的student表#xff0c;编写Python程序完成如下操作#xff1a;
#xff08;1#xff09;读取student表的数据内容#xff0c;将其转为JSON格式#xff0c;发送给Kafka#xff1b;
创建Student表的SQL语句如下#xff1a;
create table student(
sno ch…
根据上面给出的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_serverslocalhost:9092,value_serializerlambda v:json.dumps(v).encode(utf-8))connectpymysql.Connect(hostlocalhost,port3306,userroot,passwd123456,dbzhangna,charsetutf8
)
cursorconnect.cursor()
sqlselect sno,sname,ssex,sage from student;
cursor.execute(sql)
datacursor.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_idNone,auto_offset_resetearliest)
for msg in consumer:msg1str(msg.value,encodingutf-8)datajson.loads(msg1)print(data) 终于出来了出错的原因是encoding,我写成了encodings的缘故
为什么我会出现两条重复记录原因是我生产者程序运行了多次生产者多运行一次消费者程序就会多一次查询
