Kafka与MySQL的组合使用

1. 根据上面给出的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 KafkaProducerimport jsonimport 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 KafkaConsumerimport jsonimport 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的缘故

为什么我会出现两条重复记录，原因是我生产者程序运行了多次，生产者多运行一次，消费者程序就会多一次查询