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(发邮件时,请将#更换为@)进行举报,并提供相关证据,一经查实,本站将立刻删除涉嫌侵权内容。