《數(shù)據(jù)采集與處理技術》課件-4.7 Kafka與MySQL的組合使用

第4章分布式消息系統(tǒng)Kafka目

錄4.1Kafka簡介4.2Kafka在大數(shù)據(jù)生態(tài)系統(tǒng)中的作用4.3Kafka與Flume的區(qū)別與聯(lián)系4.4Kafka相關概念4.5Kafka的安裝和使用4.6使用Python操作Kafka4.7Kafka與MySQL的組合使用4.7Kafka與MySQL的組合使用4.7Kafka與MySQL的組合使用這里通過一個實例來演示Kafka與MySQL的組合使用。需要完成的任務是，把JSON格式數(shù)據(jù)放入Kafka發(fā)送出去，然后，再從Kafka中獲取到JSON格式數(shù)據(jù)，對其進行解析并寫入到MySQL數(shù)據(jù)庫。請參照第2章的內(nèi)容完成MySQL數(shù)據(jù)庫的安裝，并學習其使用方法。4.7Kafka與MySQL的組合使用編寫一個生產(chǎn)者程序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)#發(fā)送的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")#字節(jié)數(shù)組轉(zhuǎn)成字符串dict=json.loads(msg1)#字符串轉(zhuǎn)換成字典#連接數(shù)據(jù)庫connect=pymysql.Connect(host='localhost',port=3306,user='root',#數(shù)據(jù)庫用戶名passwd='123456',#密碼db='school',charset='utf8')

4.7Kafka與MySQL的組合使用#獲取游標cursor=connect.cursor()

#插入數(shù)據(jù)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('成功插入數(shù)據(jù)')

#關閉數(shù)據(jù)庫連接connect.close()4.7Kafka與MySQL的組合使用在Windows系統(tǒng)中啟動MySQL服務，然后，打開MySQL數(shù)據(jù)庫的命令行界面，輸入如下SQL語句創(chuàng)建數(shù)據(jù)庫school：mysql>CREATEDATABASEschool;創(chuàng)建好數(shù)據(jù)庫school以后，可以使用如下SQL語句打開數(shù)據(jù)庫：mysql>USEschool;使用如下SQL語句創(chuàng)建一個表student：mysql>CREATETABLEstudent(->snochar(5),->snamechar(10),->ssexchar(2),->sageint);4.7Kafka與MySQL的組合使用使用如下SQL語句查看已經(jīng)創(chuàng)建的表：mysql>SHOWTABLES;在Windows系統(tǒng)中啟動Zookeeper服務和Kafka服務，然后，先執(zhí)行生產(chǎn)者程序producer_json.py，再執(zhí)行消費者程序consumer_json.py，執(zhí)行成功以后，使用如下命令查看MySQL數(shù)據(jù)庫中新插入的記錄：mysql>SELECT*FROMstudent;可以看到，一條記錄已經(jīng)被成功地插入到了MySQL數(shù)據(jù)庫。4.8本章小結Kafka是一個分布式、分區(qū)的、多副本的、多訂閱者、基于Zookeeper協(xié)調(diào)的分布式日志系統(tǒng)，主要應用場景是日志收集系統(tǒng)和消息系統(tǒng)。LinkedIn于2010年把Kafka貢獻給了Apache基金會并成為頂級開源項目。Kafka能夠以時間復雜度為O(1)的方式提供消息持久化能力，即使對TB級以上數(shù)據(jù)也能保證常數(shù)時間的訪問性能。