python 消费 kafka 数据教程

Python  /  管理员 发布于 7年前   160

1.安装python模块

pip install --user kafka-python==1.4.3 

如果报错压缩相关的错尝试安装下面的依赖

yum install snappy-develyum install lz4-develpip install python-snappypip install lz4

2.生产者

#!/usr/bin/env python# coding : utf-8from kafka import KafkaProducerimport jsondef kafkaProducer():  producer = KafkaProducer(bootstrap_servers='ip:9092',value_serializer=lambda v: json.dumps(v).encode('utf-8'))  producer.send('world', {'key1': 'value1'})if __name__ == '__main__':  kafkaProducer()

2.消费者

from kafka import KafkaConsumerfrom kafka.structs import TopicPartitionimport timeimport clickimport ConfigParserimport jsonimport threadingimport datetimeimport schedconfig = ConfigParser.ConfigParser()config.read("amon.ini")@click.group()def cli():  [email protected]()@click.option('--topic',type=str)@click.option('--offset', type=click.Choice(['smallest', 'earliest', 'largest']))@click.option("--group",type=str)def client(topic,offset,group):  click.echo(topic)  consumer = KafkaConsumer(topic,   bootstrap_servers=config.get("KAFKA", "Broker_Servers").split(","),   group_id=group,   auto_offset_reset=offset)  for message in consumer:    click.echo(message.value)    # click.echo("%d:%d: key=%s value=%s" % (message.partition,    #          message.offset, message.key,    #          message.value))if __name__ == '__main__':  cli()

3.多线程消费

#coding:utf-8import threadingimport osimport sysfrom kafka import KafkaConsumer, TopicPartition, OffsetAndMetadatafrom collections import OrderedDictthreads = []class MyThread(threading.Thread):  def __init__(self, thread_name, topic, partition):    threading.Thread.__init__(self)    self.thread_name = thread_name    self.partition = partition    self.topic = topic  def run(self):    print("Starting " + self.name)    Consumer(self.thread_name, self.topic, self.partition)  def stop(self):    sys.exit()def Consumer(thread_name, topic, partition):  broker_list = 'ip1:9092,ip2:9092'  '''  fetch_min_bytes（int） - 服务器为获取请求而返回的最小数据量，否则请等待  fetch_max_wait_ms（int） - 如果没有足够的数据立即满足fetch_min_bytes给出的要求，服务器在回应提取请求之前将阻塞的最大时间量（以毫秒为单位）  fetch_max_bytes（int） - 服务器应为获取请求返回的最大数据量。这不是绝对最大值，如果获取的第一个非空分区中的第一条消息大于此值，  则仍将返回消息以确保消费者可以取得进展。注意：使用者并行执行对多个代理的提取，因此内存使用将取决于包含该主题分区的代理的数量。  支持的Kafka版本> = 0.10.1.0。默认值：52428800（50 MB）。  enable_auto_commit（bool） - 如果为True，则消费者的偏移量将在后台定期提交。默认值：True。  max_poll_records（int） - 单次调用中返回的最大记录数poll()。默认值：500  max_poll_interval_ms（int） - poll()使用使用者组管理时的调用之间的最大延迟 。这为消费者在获取更多记录之前可以闲置的时间量设置了上限。    如果 poll()在此超时到期之前未调用，则认为使用者失败，并且该组将重新平衡以便将分区重新分配给另一个成员。默认300000  '''  consumer = KafkaConsumer(bootstrap_servers=broker_list,   group_id="test000001",   client_id=thread_name,   enable_auto_commit=False,   fetch_min_bytes=1024 * 1024, # 1M   # fetch_max_bytes=1024 * 1024 * 1024 * 10,   fetch_max_wait_ms=60000, # 30s   request_timeout_ms=305000,   # consumer_timeout_ms=1,   # max_poll_records=5000,   )  # 设置topic partition  tp = TopicPartition(topic, partition)  # 分配该消费者的TopicPartition，也就是topic和partition，根据参数，每个线程消费者消费一个分区  consumer.assign([tp])  #获取上次消费的最大偏移量  offset = consumer.end_offsets([tp])[tp]  print(thread_name, tp, offset)  # 设置消费的偏移量  consumer.seek(tp, offset)  print u"程序首次运行\t线程:", thread_name, u"分区:", partition, u"偏移量:", offset, u"\t开始消费..."  num = 0 # 记录该消费者消费次数  while True:    msg = consumer.poll(timeout_ms=60000)    end_offset = consumer.end_offsets([tp])[tp]    '''可以自己记录控制消费'''    print u'已保存的偏移量', consumer.committed(tp), u'最新偏移量，', end_offset    if len(msg) > 0:      print u"线程:", thread_name, u"分区:", partition, u"最大偏移量:", end_offset, u"有无数据,", len(msg)      lines = 0      for data in msg.values():        for line in data:          print line          lines += 1        '''        do something        '''      # 线程此批次消息条数      print(thread_name, "lines", lines)      if True:        # 可以自己保存在各topic, partition的偏移量        # 手动提交偏移量 offsets格式：{TopicPartition:OffsetAndMetadata(offset_num,None)}        consumer.commit(offsets={tp: (OffsetAndMetadata(end_offset, None))})        if True == 0:          # 系统退出？这个还没试          os.exit()          '''          sys.exit()  只能退出该线程，也就是说其它两个线程正常运行，主程序不退出          '''      else:        os.exit()    else:      print thread_name, '没有数据'    num += 1    print thread_name, "第", num, "次"if __name__ == '__main__':  try:    t1 = MyThread("Thread-0", "test", 0)    threads.append(t1)    t2 = MyThread("Thread-1", "test", 1)    threads.append(t2)    t3 = MyThread("Thread-2", "test", 2)    threads.append(t3)    for t in threads:      t.start()    for t in threads:      t.join()    print("exit program with 0")  except:    print("Error: failed to run consumer program")

参考：https://kafka-python.readthedocs.io/en/master/index.html

https:///article/176911.htm

以上这篇python 消费 kafka 数据教程就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持。