kafka-python 獲取topic lag值方式
更新時間：2019年12月23日 09:26:33 作者：conghui_
今天小編就為大家分享一篇kafka-python 獲取topic lag值方式，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
說真，這個問題看上去很簡單，但“得益”與kafka-python神奇的文檔，真的不算簡單，反正我是搜了半天還看了半天源碼。
直接上代碼吧
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)容了，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關(guān)文章
這篇文章主要介紹了Python 如何在局部變量域中執(zhí)行代碼，幫助大家更好的理解和學習Python，感興趣的朋友可以了解下2020-08-08
pytest是python的單測框架，使用靈活，插件豐富，本文給大家分享python單測框架之pytest常見用法包括pytest常見的插件，感興趣的朋友跟隨小編一起看看吧2021-08-08
python輸出結(jié)果刷新及進度條的實現(xiàn)操作
這篇文章主要介紹了python輸出結(jié)果刷新及進度條的實現(xiàn)操作，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-07-07
python畫圖時linestyle,color和loc參數(shù)的設置方式
這篇文章主要介紹了python畫圖時linestyle，color和loc參數(shù)的設置方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-11-11
這篇文章主要介紹了Python中Django的路由配置詳解,Python下有許多款不同的?Web?框架,Django是重量級選手中最有代表性的一位,許多成功的網(wǎng)站和APP都基于Django,需要的朋友可以參考下2023-07-07
Python實現(xiàn)將MongoDB中的數(shù)據(jù)導入到MySQL
這篇文章主要為大家詳細介紹了如何通過Python封裝一個將?MongoDB?中的數(shù)據(jù)導入到?MySQL?中的?Python?工具類?MongoToMysql，感興趣的可以了解一下2023-05-05
nginx搭建基于python的web環(huán)境的實現(xiàn)步驟
這篇文章主要介紹了nginx搭建基于python的web環(huán)境的實現(xiàn)步驟，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-01-01
python處理自動化任務之同時批量修改word里面的內(nèi)容的方法
在本篇文章里小編給各位整理的是一篇關(guān)于利用python處理自動化任務之同時批量修改word里面的內(nèi)容的文章，需要的可以參考學習下。2019-08-08
python驗證公網(wǎng)ip與內(nèi)網(wǎng)ip的實現(xiàn)示例
本文主要介紹了python驗證公網(wǎng)ip與內(nèi)網(wǎng)ip的實現(xiàn)示例，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-07-07
詳細解讀tornado協(xié)程(coroutine)原理
這篇文章主要介紹了詳細解讀tornado協(xié)程(coroutine)原理，涉及協(xié)程定義，生成器和yield語義，F(xiàn)uture對象，ioloop對象，函數(shù)裝飾器coroutine等相關(guān)內(nèi)容，具有一定借鑒價值,需要的朋友可以參考下2018-01-01