小编use*_*620的帖子

我正在使用下面的代码来阅读主题的消息。我面临两个问题。每当我启动消费者时，它正在读取队列中的所有消息？如何只阅读未读邮件？

from kafka import KafkaConsumer


consumer = KafkaConsumer('my-topic',
                         group_id='my-group',
                         bootstrap_servers=['localhost:9092'])
for message in consumer:
    consumer.commit() 
    # message value and key are raw bytes -- decode if necessary!
    # e.g., for unicode: `message.value.decode('utf-8')`
    print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
                                          message.offset, message.key,
                                          message.value))

在HBase表中,我有一个包含3列的列族addr:addr:city,addr:state,addr:zip.我想找到city = Chicago的所有行.任何想要使用过滤器的想法？什么是java语法？