kafka-python 获取topic lag值方式
说真,这个问题看上去很简单,但“得益”与kafka-python神奇的文档,真的不算简单,反正我是搜了半天还看了半天源码。
直接上代码吧
fromkafkaimportSimpleClient,KafkaConsumer fromkafka.commonimportOffsetRequestPayload,TopicPartition defget_topic_offset(brokers,topic): """ 获取一个topic的offset值的和 """ client=SimpleClient(brokers) partitions=client.topic_partitions[topic] offset_requests=[OffsetRequestPayload(topic,p,-1,1)forpinpartitions.keys()] offsets_responses=client.send_offset_request(offset_requests) returnsum([r.offsets[0]forrinoffsets_responses]) defget_group_offset(brokers,group_id,topic): """ 获取一个topic特定group已经消费的offset值的和 """ consumer=KafkaConsumer(bootstrap_servers=brokers, group_id=group_id, ) pts=[TopicPartition(topic=topic,partition=i)foriin consumer.partitions_for_topic(topic)] result=consumer._coordinator.fetch_committed_offsets(pts) returnsum([r.offsetforrinresult.values()]) if__name__=='__main__': topic_offset=get_topic_offset("brokers","topic") group_offset=get_group_offset("brokers","group_id","topic") lag=topic_offset-group_offset
以上这篇kafka-python获取topiclag值方式就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持毛票票。
声明:本文内容来源于网络,版权归原作者所有,内容由互联网用户自发贡献自行上传,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任。如果您发现有涉嫌版权的内容,欢迎发送邮件至:czq8825#qq.com(发邮件时,请将#更换为@)进行举报,并提供相关证据,一经查实,本站将立刻删除涉嫌侵权内容。