python - 如何使用pykafka consumer進行數據處理并保存？

問題描述

使用本地kafka bin/kafka-console-producer.sh --broker-list kafkaIP:port --topic topicName創建命令行生產數據，然后打開python

from pykafka import KafkaClientclient = KafkaClient(hosts='192.168.x.x:9092')topic = client.topics[’wr_test’]consumer = topic.get_balanced_consumer(consumer_group=’test-consumer-group’,auto_commit_enable=True,zookeeper_connect=’192.168.x.x:2121’)

然后自己編寫了簡單的一套處理函數，從外部引用。將數據處理后存入elasticsearch 或者 數據庫比如for msg in consumer:

if msg is not None: 外部引入的處理函數(msg.value)

在python命令行for msg in consumer:

print msg.offset, msg.value

這時候使用生產者敲入一些數據，在消費端就會就會立即打印出來但是寫成py文件之后，每次運行只會處理最近的生產的一次內容，在生產者中再進行輸入一些內容，py文件就不會再進行數據處理了。所以向問下如何編寫能運行后能一直對消費者數據進行處理的函數？要注意哪些地方？

另外，get_balanced_consumer的方法，是連接zookeeper消費使用topic.get_simple_consumer是直接消費kafka，使用這種方式就提示No handler for...的錯誤

還有一個疑問，就是實際生產環境日志產生量很快，應該如何編寫一個多線程處理方法？

問題解答

在別人的博客看到一種替代的解決方案http://www.cnblogs.com/castle...從consumer中將msg.value讀取到一個列表當中，然后從列表中讀取數據進行數據處理，當這個流程結束后，再把列表中獲取的數據pop掉。另外也要用try: ... except :... continue