如下所示:
1
2
|
from kafka import KafkaClient
from kafka.producer import SimpleProducer
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
def send_data_2_kafka(datas):
'''
向kafka解析队列发送数据
'''
client = KafkaClient(hosts = KAFKABROKER.split( "," ), timeout = 30 )
producer = SimpleProducer(client, async = False )
curcount = len (datas) / PARTNUM
for i in range ( 0 , PARTNUM):
start = i * curcount
if i ! = PARTNUM - 1 :
end = (i + 1 ) * curcount
curdata = datas[start:end]
producer.send_messages(TOPICNAME, * curdata)
else :
curdata = datas[start:]
producer.send_messages(TOPICNAME, * curdata)
producer.stop()
client.close()
|
其中PARTNUM为topic的partition的数目,这样保证批量发送的数据均匀的落在kafka的partition中。
以上这篇kafka-python批量发送数据的实例就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持服务器之家。
原文链接:https://blog.csdn.net/rongyongfeikai2/article/details/54576340