如何ConsumerRecord

问题描述：

取一个字段我写了一个python脚本：如何ConsumerRecord

#!/usr/bin/env python 
from kafka import KafkaConsumer 
consumer = KafkaConsumer('dimon_tcpdump',group_id='zhg_group',bootstrap_servers='192.168.100.9:9092') 
for msg in consumer: 
    print msg 
    # process mes here

味精，输出类似：

ConsumerRecord(topic=u'ditopic', partition=0, offset=6280, timestamp=None, timestamp_type=None, key=None, value='myvalue')

我知道这个输出是namedtuple形式。

我的问题是：如何获得ConsumerRecord的特定字段？例如，我想将值字符串分配给一个变量。

我猜你可以叫'msg.value'提取场'value' ？ –

>>> msg = r'''ConsumerRecord（topic = u'ditopic'，partition = 0，offset = 6280，timestamp = None，timestamp_type = None，key = None，value ='myvalue'）''' > >> msg “ConsumerRecord（topic = u'ditopic'，partition = 0，offset = 6280，timestamp = None，timestamp_type = None，key = None，value ='myvalue'）”msg.value Traceback （最近呼叫最后）：文件“”，第1行，在 AttributeError：'str'对象没有属性'值' >>> – zhenghuagui

它应该是'msg.value' - 不知道为什么它不适合你。请参阅自述文件中的代码示例https://github.com/dpkp/kafka-python#kafkaconsumer –

答

它可能与您的反序列化数据有关。例如，如果你想从msg获取一些JSON。你会初始化Consumer有：

value_deserializer=lambda m: json.loads(m.decode('utf-8'))

所以您的代码会是这个样子：

#!/usr/bin/env python 
from kafka import KafkaConsumer 
consumer = KafkaConsumer(
    'dimon_tcpdump', 
    group_id='zhg_group', 
    value_deserializer=lambda m: json.loads(m.decode('utf-8')), 
    bootstrap_servers='192.168.100.9:9092' 
    ) 
for msg in consumer: 
    print msg.value 
    # process mes here

相关推荐