溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
背景：
為了滿足各個平臺間數據的傳輸，以及能確保歷史性和實時性。先選用kafka作為不同平臺數據傳輸的中轉站，來滿足我們對跨平臺數據發送與接收的需要。
kafka簡介：
Kafka is a distributed,partitioned,replicated commit logservice。它提供了類似于JMS的特性，但是在設計實現上完全不同，此外它并不是JMS規范的實現。kafka對消息保存時根據Topic進行歸類，發送消息者成為Producer,消息接受者成為Consumer,此外kafka集群有多個kafka實例組成，每個實例(server)成為broker。無論是kafka集群，還是producer和consumer都依賴于zookeeper來保證系統可用性集群保存一些meta信息。
總之：kafka做為中轉站有以下功能：
1.生產者（產生數據或者說是從外部接收數據）
2.消費著（將接收到的數據轉花為自己所需用的格式）
環境：
1.python3.5.x
2.kafka1.4.3
3.pandas
準備開始：
1.kafka的安裝
pip install kafka-python
2.檢驗kafka是否安裝成功
3.pandas的安裝
pip install pandas
4.kafka數據的傳輸
直接擼代碼：
# -*- coding: utf-8 -*- ''' @author: 真夢行路 @file: kafka.py @time: 2018/9/3 10:20 ''' import sys import json import pandas as pd import os from kafka import KafkaProducer from kafka import KafkaConsumer from kafka.errors import KafkaError KAFAKA_HOST = "xxx.xxx.x.xxx" #服務器端口地址 KAFAKA_PORT = 9092 #端口號 KAFAKA_TOPIC = "topic0" #topic data=pd.read_csv(os.getcwd()+'\\data\\1.csv') key_value=data.to_json() class Kafka_producer(): ''' 生產模塊：根據不同的key，區分消息 ''' def __init__(self, kafkahost, kafkaport, kafkatopic, key): self.kafkaHost = kafkahost self.kafkaPort = kafkaport self.kafkatopic = kafkatopic self.key = key self.producer = KafkaProducer(bootstrap_servers='{kafka_host}:{kafka_port}'.format( kafka_host=self.kafkaHost, kafka_port=self.kafkaPort) ) def sendjsondata(self, params): try: parmas_message = params #注意dumps producer = self.producer producer.send(self.kafkatopic, key=self.key, value=parmas_message.encode('utf-8')) producer.flush() except KafkaError as e: print(e) class Kafka_consumer(): def __init__(self, kafkahost, kafkaport, kafkatopic, groupid,key): self.kafkaHost = kafkahost self.kafkaPort = kafkaport self.kafkatopic = kafkatopic self.groupid = groupid self.key = key self.consumer = KafkaConsumer(self.kafkatopic, group_id=self.groupid, bootstrap_servers='{kafka_host}:{kafka_port}'.format( kafka_host=self.kafkaHost, kafka_port=self.kafkaPort) ) def consume_data(self): try: for message in self.consumer: yield message except KeyboardInterrupt as e: print(e) def sortedDictValues(adict): items = adict.items() items=sorted(items,reverse=False) return [value for key, value in items] def main(xtype, group, key): ''' 測試consumer和producer ''' if xtype == "p": # 生產模塊 producer = Kafka_producer(KAFAKA_HOST, KAFAKA_PORT, KAFAKA_TOPIC, key) print("===========> producer:", producer) params =key_value producer.sendjsondata(params) if xtype == 'c': # 消費模塊 consumer = Kafka_consumer(KAFAKA_HOST, KAFAKA_PORT, KAFAKA_TOPIC, group,key) print("===========> consumer:", consumer) message = consumer.consume_data() for msg in message: msg=msg.value.decode('utf-8') python_data=json.loads(msg) ##這是一個字典 key_list=list(python_data) test_data=pd.DataFrame() for index in key_list: print(index) if index=='Month': a1=python_data[index] data1 = sortedDictValues(a1) test_data[index]=data1 else: a2 = python_data[index] data2 = sortedDictValues(a2) test_data[index] = data2 print(test_data) # print('value---------------->', python_data) # print('msg---------------->', msg) # print('key---------------->', msg.kry) # print('offset---------------->', msg.offset) if __name__ == '__main__': main(xtype='p',group='py_test',key=None) main(xtype='c',group='py_test',key=None)
數據1.csv如下所示：
幾點注意：
1、一定要有一個服務器的端口地址，不要用本機的ip或者亂寫一個ip不然程序會報錯。（我開始就是拿本機ip懟了半天，總是報錯）
2、注意數據的傳輸格式以及編碼問題(二進制傳輸)，數據先轉成json數據格式傳輸，然后將json格式轉為需要格式。（不是json格式的注意dumps）
例中，dataframe->json->dataframe
3、例中dict轉dataframe,也可以用簡單方法直接轉。
eg： type(data) ==>dict,data=pd.Dateframe(data)
以上這篇在python環境下運用kafka對數據進行實時傳輸的方法就是小編分享給大家的全部內容了，希望能給大家一個參考，也希望大家多多支持億速云。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
