《数据采集与处理技术》课件——4.7 Kafka与MySQL的组合使用.pptxVIP

第4章分布式消息系统Kafka

目录4.1Kafka简介4.2Kafka在大数据生态系统中的作用4.3Kafka与Flume的区别与联系4.4Kafka相关概念4.5Kafka的安装和使用4.6使用Python操作Kafka4.7Kafka与MySQL的组合使用

4.7Kafka与MySQL的组合使用

4.7Kafka与MySQL的组合使用这里通过一个实例来演示Kafka与MySQL的组合使用。需要完成的任务是，把JSON格式数据放入Kafka发送出去，然后，再从Kafka中获取到JSON格式数据，对其进行解析并写入到MySQL数据库。请参照第2章的内容完成MySQL数据库的安装，并学习其使用方法。

4.7Kafka与MySQL的组合使用编写一个生产者程序producer_json.py：#producer_json.pyfromkafkaimportKafkaProducerimportjson?producer=KafkaProducer(bootstrap_servers=localhost:9092,value_serializer=lambdav:json.dumps(v).encode(utf-8))#连接kafka?data={sno:95001,name:John,sex:M,age:23}?producer.send(json_topic,data)#发送的topic为json_topicproducer.close()

4.7Kafka与MySQL的组合使用编写一个消费者程序consumer_json.py：#consumer_json.pyfromkafkaimportKafkaConsumerimportjsonimportpymysql.cursors?consumer=KafkaConsumer(json_topic,bootstrap_servers=[localhost:9092],group_id=None,auto_offset_reset=earliest)formsginconsumer:msg1=str(msg.value,encoding=utf-8)#字节数组转成字符串dict=json.loads(msg1)#字符串转换成字典#连接数据库connect=pymysql.Connect(host=localhost,port=3306,user=root,#数据库用户名passwd=123456,#密码db=school,charset=utf8)?

4.7Kafka与MySQL的组合使用#获取游标cursor=connect.cursor()?#插入数据sql=INSERTINTOstudent(sno,sname,ssex,sage)VALUES(%s,%s,%s,%d)data=(dict[sno],dict[name],dict[sex],dict[age])cursor.execute(sql%data)mit()print(成功插入数据)?#关闭数据库连接connect.close()

4.7Kafka与MySQL的组合使用在Windows系统中启动MySQL服务，然后，打开MySQL数据库的命令行界面，输入如下SQL语句创建数据库school：mysqlCREATEDATABASEschool;创建好数据库school以后，可以使用如下SQL语句打开数据库：mysqlUSEschool;使用如下SQL语句创建一个表student：mysqlCREATETABLEstudent(-snochar(5),-snamechar(10),-ssexchar(2),-sageint);

4.7Kafka与MySQL的组合使用使用如下SQL语句查看已经创建的表：mysqlSHOWTABLES;在Windows系统中启动Zookeeper服务和Kafka服务，然后，先执行生产者程序producer_json.py，再执行消费者程序consumer_json.py，执行成功以后，使用如下命令查看MySQL数据库中新插入的记录：mysqlSELECT*FROMstudent;可以看到，一条记录已经被成功地插入到了MySQL数据库。

4.8