from kafka import KafkaClient
from import SimpleProducer
def send_data_2_kafka(datas):
'''
向kafka解析队列发送数据
'''
client = KafkaClient(hosts=(","), 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)
()
()
其中PARTNUM为topic的partition的数目,这样保证批量发送的数据均匀的落在kafka的partition中。