Kafka与MySQL的组合使用-Toy模板网

这篇具有很好参考价值的文章主要介绍了Kafka与MySQL的组合使用。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

根据上面给出的student表，编写Python程序完成如下操作：

（1）读取student表的数据内容，将其转为JSON格式，发送给Kafka；

创建Student表的SQL语句如下：

create table student(
sno char(5),
sname char(10),
ssex char(2),
sage int
);

kafka连接mysql,大数据采集与预处理,mysql,数据库,kafka

向student表中插入两条记录的SQL语句如下：

insert into student values(‘95001’,’John’,’M’,23);
insert into student values(‘95002’,’Tom’,’M’,23);

kafka连接mysql,大数据采集与预处理,mysql,数据库,kafka

启动zookeeper和kafka的服务

kafka连接mysql,大数据采集与预处理,mysql,数据库,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()

kafka连接mysql,大数据采集与预处理,mysql,数据库,kafka

（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)

kafka连接mysql,大数据采集与预处理,mysql,数据库,kafka