您好,登錄后才能下訂單哦!
這篇文章主要介紹kafka-python怎么獲取topic lag值方式,文中介紹的非常詳細,具有一定的參考價值,感興趣的小伙伴們一定要看完!
直接上代碼吧
from kafka import SimpleClient, KafkaConsumer from kafka.common import OffsetRequestPayload, TopicPartition def get_topic_offset(brokers, topic): """ 獲取一個topic的offset值的和 """ client = SimpleClient(brokers) partitions = client.topic_partitions[topic] offset_requests = [OffsetRequestPayload(topic, p, -1, 1) for p in partitions.keys()] offsets_responses = client.send_offset_request(offset_requests) return sum([r.offsets[0] for r in offsets_responses]) def get_group_offset(brokers, group_id, topic): """ 獲取一個topic特定group已經(jīng)消費的offset值的和 """ consumer = KafkaConsumer(bootstrap_servers=brokers, group_id=group_id, ) pts = [TopicPartition(topic=topic, partition=i) for i in consumer.partitions_for_topic(topic)] result = consumer._coordinator.fetch_committed_offsets(pts) return sum([r.offset for r in result.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怎么獲取topic lag值方式”這篇文章的所有內(nèi)容,感謝各位的閱讀!希望分享的內(nèi)容對大家有幫助,更多相關(guān)知識,歡迎關(guān)注億速云行業(yè)資訊頻道!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。