我尝试使用Kafka Python作为流式传输数据的一种非常基本的方式,在两个jupyter笔记本中设置一个KafkaProducer和一个KafkaConsumer,类似于本文:https://dorianbg.wordpress.com/2017/11/11/ingesting-realtime-tweets-using-apache-kafka-tweepy-and-python/
设置我的制作人:
import time
!pip install kafka-python
from kafka import KafkaConsumer, KafkaProducer
producer = KafkaProducer(bootstrap_servers=['localhost:9092'], api_version=(0, 10, 1), max_block_ms = 300000, api_version_auto_timeout_ms=300000)
topic_name = 'test'
def send_data(interval):
while True:
producer.send(topic_name, str(1))
time.sleep(interval)
设置我的消费者:
^{pr2}$当我在单独的笔记本中运行这些程序时,我在producer笔记本中收到以下消息:
KafkaTimeoutError: KafkaTimeoutError: Failed to update metadata after 300.0 secs.
我很清楚这是一个以前被问过的问题(KafkaTimeoutError('Failed to update metadata after 60.0 secs.'),https://github.com/dpkp/kafka-python/issues/721,等等),但是这些情况与我的不同,或者提供的答案不能解决我的问题。在
谢谢!在
目前没有回答
相关问题 更多 >
编程相关推荐