kafka python 指定分區消費 與 offset

 

指定offset:

#pip install kafka-python
import gzip
from kafka import KafkaConsumer
from kafka import TopicPartition

consumer = KafkaConsumer(bootstrap_servers='127.0.0.1:9092')
partition = TopicPartition('mytopic', 0)
start = 8833
end = 8835 consumer.assign([partition]) consumer.seek(partition, start)

i=start
for msg in consumer:
    if msg.offset > end:
        break
    else:
        print msg
        try:
            outF = file( '/data/app/tar/' + str(i) + ".gz", 'wb')
            outF.write(msg.value)
            outF.close()
            i+=1
            print i
        except Exception,e:
            print e

 

 

 

指定分區：

通過assign、subscribe兩者之一為消費者設置消費的主題

consumer = KafkaConsumer(bootstrap_servers=['127.0.0.1:9092'],

                         auto_offset_reset='latest',

                         enable_auto_commit=True, # 自動提交消費數據的offset

                         consumer_timeout_ms= 10000, # 如果1秒內kafka中沒有可供消費的數據，自動退出

                         value_deserializer=lambda m: json.loads(m.decode('ascii')), #消費json 格式的消息

                         client_id='consumer-python3'

                         )

 

 

# consumer.assign([TopicPartition('MY_TOPIC1', 0)])

# msg = next(consumer)

# print(msg)

 

consumer.subscribe('MY_TOPIC1')

for msg in consumer:

    print (msg)

 

 

API及常用參數說明：

class kafka.KafkaConsumer(*topics, **configs)

*topics (str) – 可選，設置需要訂閱的topic，如果未設置，需要在消費記錄前調用subscribe或者assign。

 

client_id (str) – 客戶端名稱，默認值: ‘kafka-python-{version}’

 

group_id (str or None) – 消費組名稱。如果為None，則通過group coordinator auto-partition分區分配，offset提交被禁用。默認為None

 

auto_offset_reset (str) – 重置offset策略: 'earliest'將移動到最老的可用消息, 'latest'將移動到最近消息。 設置為其它任何值將拋出異常。默認值：'latest'。

 

enable_auto_commit (bool) –  如果為True，將自動定時提交消費者offset。默認為True。

 

auto_commit_interval_ms (int) – 自動提交offset之間的間隔毫秒數。如果enable_auto_commit 為true，默認值為： 5000。

 

value_deserializer(可調用對象) - 攜帶原始消息value並返回反序列化后的value

 

subscribe(topics=(), pattern=None, listener=None)

訂閱需要的主題

topics (list) – 需要訂閱的主題列表

pattern (str) – 用於匹配可用主題的模式，即正則表達式。注意：必須提供topics、pattern兩者參數之一，但不能同時提供兩者。

 

metrics(raw=False)

獲取消費者性能指標。

 

參考API：https://kafka-python.readthedocs.io/en/master/apidoc/KafkaConsumer.html

原文強參考：https://www.cnblogs.com/shouke/p/10463377.html