python批量向kafka塞數據

python批量向kafka塞數據

 

from kafka import KafkaClient
from kafka.producer import SimpleProducer


def send_data_2_kafka(topic, datas):
    '''
        向kafka解析隊列發送數據
    '''
    PARTNUM = 100
    TOPICNAME = topic
    KAFKABROKER = ["192.168.2.110:9092"]
    client = KafkaClient(hosts=KAFKABROKER, timeout=30)
    producer = SimpleProducer(client, async_send=False)
    curcount = int(len(datas)//PARTNUM)
    for i in range(curcount):
        start = i*PARTNUM
        if i != curcount - 1:
            end = (i+1)*PARTNUM
            curdata = datas[start:end]
            future = producer.send_messages(TOPICNAME, *curdata)
            print(future)
        else:
            curdata = datas[start:]
            future = producer.send_messages(TOPICNAME, *curdata)
            print(future)
        
    producer.stop()
    client.close()