久久久久久免费视频,激情特色在线

腳本之家服務(wù)器常用軟件

快捷導(dǎo)航

在python環(huán)境下運用kafka對數(shù)據(jù)進行實時傳輸?shù)姆椒?/h1>

更新時間：2018年12月27日 10:37:06 作者：真夢行路

今天小編就為大家分享一篇在python環(huán)境下運用kafka對數(shù)據(jù)進行實時傳輸?shù)姆椒?，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧

背景：

為了滿足各個平臺間數(shù)據(jù)的傳輸，以及能確保歷史性和實時性。先選用kafka作為不同平臺數(shù)據(jù)傳輸?shù)闹修D(zhuǎn)站，來滿足我們對跨平臺數(shù)據(jù)發(fā)送與接收的需要。

kafka簡介：

Kafka is a distributed,partitioned,replicated commit logservice。它提供了類似于JMS的特性，但是在設(shè)計實現(xiàn)上完全不同，此外它并不是JMS規(guī)范的實現(xiàn)。kafka對消息保存時根據(jù)Topic進行歸類，發(fā)送消息者成為Producer,消息接受者成為Consumer,此外kafka集群有多個kafka實例組成，每個實例(server)成為broker。無論是kafka集群，還是producer和consumer都依賴于zookeeper來保證系統(tǒng)可用性集群保存一些meta信息。

總之：kafka做為中轉(zhuǎn)站有以下功能：

1.生產(chǎn)者（產(chǎn)生數(shù)據(jù)或者說是從外部接收數(shù)據(jù)）

2.消費著（將接收到的數(shù)據(jù)轉(zhuǎn)花為自己所需用的格式）

環(huán)境：

1.python3.5.x

2.kafka1.4.3

3.pandas

準(zhǔn)備開始：

1.kafka的安裝

pip install kafka-python

python環(huán)境下運用kafka對數(shù)據(jù)進行傳輸

2.檢驗kafka是否安裝成功

python環(huán)境下運用kafka對數(shù)據(jù)進行傳輸

3.pandas的安裝

pip install pandas

4.kafka數(shù)據(jù)的傳輸

直接擼代碼：

# -*- 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" #服務(wù)器端口地址
KAFAKA_PORT = 9092    #端口號
KAFAKA_TOPIC = "topic0"  #topic
 
data=pd.read_csv(os.getcwd()+'\\data\\1.csv')
key_value=data.to_json()
class Kafka_producer():
 '''
 生產(chǎn)模塊：根據(jù)不同的key，區(qū)分消息
 '''
 
 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":
  # 生產(chǎn)模塊
  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)

python環(huán)境下運用kafka對數(shù)據(jù)進行傳輸