如下所示:
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