Pyhone3 从Kafka拿取数据

前言

这篇博客搭配之前写的oracle，mysql数据同步使用

---

一、安装pykafka

pip install pykafka

二、使用步骤

1.列出所有topic

from pykafka import KafkaClient

client = KafkaClient(hosts="ip:9092")  #连接kafka
for topic in client.topics:
    print(topic)

2.读取数据

import json
from pykafka import KafkaClient

client = KafkaClient(hosts="ip:9092")   # 连接
topic = client.topics['test']   # 连接的topics
consumer = topic.get_simple_consumer(consumer_group="test",reset_offset_on_start=True)

for message in consumer:
    if message is not None:
        print(message.offset)   # 偏移量
        print(message.value)    # kafka数据
        try:
            data = json.loads(message.value.decode("utf-8"))['payload']['data']
            if data is not None:
                print(data)  # 拿取同步数据
        except:pass