用于R获取的Kafakacomsumer

Question

我正在研究卡夫卡。我在我的服务器上创建了kafka producer。我想从kafkaproducer中获取数据到r中的本地系统。

我在R中尝试了下面的代码：

library(rkafka)
consumer1<-rkafka.createConsumer("ipaddress:9092","mytest")

consumer11 <-  rkafka.read(consumer1)

它抛出以下错误：

[1] "Java-Object{com.musigma.consumer.MuConsumer@3349e9bb}"
Unable to connect to zookeeper server
org.I0Itec.zkclient.exception.ZkTimeoutException: Unable to connect to zookeeper server within 
timeout: 100000
at org.I0Itec.zkclient.ZkClient.connect(ZkClient.java:880)
at org.I0Itec.zkclient.ZkClient.<init>(ZkClient.java:98)
at org.I0Itec.zkclient.ZkClient.<init>(ZkClient.java:84)
at kafka.consumer.ZookeeperConsumerConnector.connectZk(ZookeeperConsumerConnector.scala:156)
at kafka.consumer.ZookeeperConsumerConnector.<init>(ZookeeperConsumerConnector.scala:114)
at kafka.javaapi.consumer.ZookeeperConsumerConnector.<init>(ZookeeperConsumerConnector.scala:65)
at kafka.javaapi.consumer.ZookeeperConsumerConnector.<init>(ZookeeperConsumerConnector.scala:67)
at kafka.consumer.Consumer$.createJavaConsumerConnector(ConsumerConnector.scala:100)
at kafka.consumer.Consumer.createJavaConsumerConnector(ConsumerConnector.scala)
at com.musigma.consumer.MuConsumer.CreateConsumer(MuConsumer.java:99)
java.lang.NullPointerException
at com.musigma.consumer.MuConsumer.startConsumer(MuConsumer.java:133)

我的zookeeper在is地址上运行成功。

Answer 1

第一个参数是Zookeeper，它在端口2181上运行

你给它提供了Kafka端口

来源- https://github.com/cran/rkafkajars/blob/master/java/com/musigma/consumer/MuConsumer.java#L87

注意:看起来这个库没有得到维护，而且使用Zookeeper与消费者连接实际上已经被弃用了，所以也许可以尝试寻找另一个库