python消費kafka數據

有兩個模塊都可以使用消費kafka數據

kafka模塊 和 pykafka模塊

注意kafka會將hosts轉換成域名的形式，注意要將hosts及域名配置到docker和主機的/etc/hosts文件中

一、kafka模塊

支持版本：

 

from kafka import KafkaConsumer
import time

kafka_hosts = ["192.168.3.145:49154"]
kafka_topic = "topic1"
consumer = KafkaConsumer(kafka_topic, group_id="test_aaa", bootstrap_servers=kafka_hosts, ssl_check_hostname=False,
                         api_version=(0, 9))  # 要設置api_version否則可能會報錯，如果沒有使用ssl認證設置為False

# 方式一：

print("t1", time.time())  # 記錄拉取時間
while True:
    print("t2", time.time())
    msg = consumer.poll(timeout_ms=100, max_records=5)  # 從kafka獲取消息，每0.1秒拉取一次，單次拉取5條
    print(len(msg))

    for i in msg.values():
        for k in i:
            print(k.offset, k.value)
    time.sleep(1)

# 方式二：

for message in consumer:
    if message is not None:
        print("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
                                             message.offset, message.key,
                                             message.value.decode()))
    else:
        print("30s內未接受到數據")

 

二、pykafka模塊

from pykafka import KafkaClient


class KafkaTest(object):
    def __init__(self, host):
        self.host = host
        self.client = KafkaClient(hosts=self.host)
        print(self.client.topics)  # 所有的topic
        print(self.client.brokers)  # kafka的所有的brokers

    def balance_consumer(self, topic, offset=0):
        """
        使用balance consumer去消費kafka
        :return:
        """

        topic = self.client.topics[topic.encode()]
        # managed=True 設置后，使用新式reblance分區方法，不需要使用zk，而False是通過zk來實現reblance的需要使用zk,必須指定 # zookeeper_connect = "zookeeperIp",consumer_group='test_group',
        consumer = topic.get_balanced_consumer(
            auto_commit_enable=True,
            managed=True,
            # managed=False,
            # zookeeper_connect="10.111.64.225:2181",
            consumer_group=b'HpDailyDataConsumerGroup',
            consumer_timeout_ms=300000
        )

        partitions = topic.partitions
        print("所有的分區：{}".format(partitions))
        earliest_offsets = topic.earliest_available_offsets()
        print("最早可用offset：{}".format(earliest_offsets))
        last_offsets = topic.latest_available_offsets()
        print("最近可用offset：{}".format(last_offsets))
        offset = consumer.held_offsets
        print("當前消費者分區offset情況：{}".format(offset))
        while True:
            msg = consumer.consume()
            if msg:
                offset = consumer.held_offsets
                print("當前位移：{}".format(offset))
                # result.append(eval(msg.value.decode()))
                print(msg.value.decode())
                consumer.commit_offsets()  # commit一下

            else:
                print("沒有數據")



if __name__ == '__main__':
    host = '192.168.3.145:49154'
    # host = 'c6140sv02:6667'
    kafka_ins = KafkaTest(host)
    topic = 'topic1'
    # topic = 'topic_jsonctr_collector_ws_data_daily'
    kafka_ins.balance_consumer(topic)