python 消费 kafka 数据教程
1.安装python模块
pipinstall--userkafka-python==1.4.3
如果报错压缩相关的错尝试安装下面的依赖
yuminstallsnappy-devel yuminstalllz4-devel pipinstallpython-snappy pipinstalllz4
2.生产者
#!/usr/bin/envpython
#coding:utf-8
fromkafkaimportKafkaProducer
importjson
defkafkaProducer():
producer=KafkaProducer(bootstrap_servers='ip:9092',value_serializer=lambdav:json.dumps(v).encode('utf-8'))
producer.send('world',{'key1':'value1'})
if__name__=='__main__':
kafkaProducer()
2.消费者
fromkafkaimportKafkaConsumer
fromkafka.structsimportTopicPartition
importtime
importclick
importConfigParser
importjson
importthreading
importdatetime
importsched
config=ConfigParser.ConfigParser()
config.read("amon.ini")
@click.group()
defcli():
pass
@cli.command()
@click.option('--topic',type=str)
@click.option('--offset',type=click.Choice(['smallest','earliest','largest']))
@click.option("--group",type=str)
defclient(topic,offset,group):
click.echo(topic)
consumer=KafkaConsumer(topic,
bootstrap_servers=config.get("KAFKA","Broker_Servers").split(","),
group_id=group,
auto_offset_reset=offset)
formessageinconsumer:
click.echo(message.value)
#click.echo("%d:%d:key=%svalue=%s"%(message.partition,
#message.offset,message.key,
#message.value))
if__name__=='__main__':
cli()
3.多线程消费
#coding:utf-8
importthreading
importos
importsys
fromkafkaimportKafkaConsumer,TopicPartition,OffsetAndMetadata
fromcollectionsimportOrderedDict
threads=[]
classMyThread(threading.Thread):
def__init__(self,thread_name,topic,partition):
threading.Thread.__init__(self)
self.thread_name=thread_name
self.partition=partition
self.topic=topic
defrun(self):
print("Starting"+self.name)
Consumer(self.thread_name,self.topic,self.partition)
defstop(self):
sys.exit()
defConsumer(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（50MB）。
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,
)
#设置topicpartition
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)
printu"程序首次运行\t线程:",thread_name,u"分区:",partition,u"偏移量:",offset,u"\t开始消费..."
num=0#记录该消费者消费次数
whileTrue:
msg=consumer.poll(timeout_ms=60000)
end_offset=consumer.end_offsets([tp])[tp]
'''可以自己记录控制消费'''
printu'已保存的偏移量',consumer.committed(tp),u'最新偏移量，',end_offset
iflen(msg)>0:
printu"线程:",thread_name,u"分区:",partition,u"最大偏移量:",end_offset,u"有无数据,",len(msg)
lines=0
fordatainmsg.values():
forlineindata:
printline
lines+=1
'''
dosomething
'''
#线程此批次消息条数
print(thread_name,"lines",lines)
ifTrue:
#可以自己保存在各topic,partition的偏移量
#手动提交偏移量offsets格式：{TopicPartition:OffsetAndMetadata(offset_num,None)}
consumer.commit(offsets={tp:(OffsetAndMetadata(end_offset,None))})
ifTrue==0:
#系统退出？这个还没试
os.exit()
'''
sys.exit()只能退出该线程，也就是说其它两个线程正常运行，主程序不退出
'''
else:
os.exit()
else:
printthread_name,'没有数据'
num+=1
printthread_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)
fortinthreads:
t.start()
fortinthreads:
t.join()
print("exitprogramwith0")
except:
print("Error:failedtorunconsumerprogram")
参考：https://kafka-python.readthedocs.io/en/master/index.html
https://www.nhooo.com/article/176911.htm
以上这篇python消费kafka数据教程就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持毛票票。
声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。