0

我的目标是从非文件源（即在程序中生成或通过 API 发送）获取数据并将其发送到 spark 流。为此，我通过基于python 的方式 KafkaProducer发送数据：

$ bin/zookeeper-server-start.sh config/zookeeper.properties &
$ bin/kafka-server-start.sh config/server.properties &
$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic my-topic
$ python 
Python 3.6.1| Anaconda custom (64-bit)
> from kafka import KafkaProducer
> import time
> producer = KafkaProducer(bootstrap_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode('utf-8'))
> producer.send(topic = 'my-topic', value = 'MESSAGE ACKNOWLEDGED', timestamp_ms = time.time())
> producer.close()
> exit()

我的问题是从消费者 shell 脚本检查主题时没有出现任何内容：

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:2181 --topic my-topic
^C$

这里有什么遗漏或错误吗？我是 spark/kafka/messaging 系统的新手，所以任何事情都会有所帮助。Kafka 版本是 0.11.0.0 (Scala 2.11)，并且没有对配置文件进行任何更改。

4

2 回答 2

1

如果您在向主题发送消息后启动消费者，消费者可能会跳过该消息，因为它将设置主题偏移量（可以被视为读取的“起点”）到主题的结尾。要更改该行为，请尝试添加--from-beginning选项：

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning

你也可以试试kafkacat，这比Kafka的控制台消费者和生产者（恕我直言）更方便。kafkacat可以使用以下命令从 Kafka 读取消息：

kafkacat -C -b 'localhost:9092' -o beginning -e -D '\n' -t 'my-topic'

希望它会有所帮助。

于 2017-07-08T21:38:15.877 回答
0

我发现了问题，value_serializer因为我没有将 json 模块导入解释器，所以默默地崩溃了。对此有两种解决方案，一种是简单地导入模块，然后您将得到"MESSAGE ACKNOWLEDGED"（带引号）返回。或者您可以value_serializer完全删除并将value在下一行中发送的字符串转换为字节字符串（即b'MESSAGE ACKNOWLEDGED'对于 Python 3），这样您就可以得到不带引号的消息。

我还将 Kafka 切换到 0.10.2.1 版（Scala 2.11），因为 Kafka-python 文档中没有确认它与 0.11.0.0 版兼容

于 2017-07-11T03:53:43.330 回答