Невозможно отправить события json tweets в тему/продюсер Kafka с помощью командной строки kafka

Question

Я создал сценарий python raw_tweets_stream.py для потоковой передачи данных twitter с помощью twitter api. Данные json из твиттера передаются производителю кафки, используя сценарий ниже.

---

`python raw_tweets_stream.py | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list localhost:2181 --topic raw_json_tweets` 

---

raw_json_tweets является Кафка тема создана для этих твитов. Питон python raw_tweets_stream.py работает отлично, но он выдает ошибку при отправке ее производителю kafka. Я использую песочницу Hortonworks HDP 2.3.1, и я убедился, что zookeeper и kafka запущены.

---

/usr/hdp/current/kafka-broker/bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic raw_json_tweets

Topic:raw_json_tweets  PartitionCount:1  ReplicationFactor:1  Configs: 
      Topic: raw_json_tweets  Partition: 0 Leader: 0  Replicas: 0  Isr: 0 

Ошибка:

[2016-08-25 22:36:26,212] ERROR Failed to send requests for topics raw_json_tweets with correlation ids in [57,64] (kafka.producer.async.DefaultEventHandler) 
[2016-08-25 22:36:26,213] ERROR Error in handling batch of 131 events (kafka.producer.async.ProducerSendThread) 
kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries. 
     at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:91) 
     at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105) 
     at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:88) 
     at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:68) 
     at scala.collection.immutable.Stream.foreach(Stream.scala:547) 
     at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67) 
     at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45) 
[2016-08-25 22:36:27,217] WARN Fetching topic metadata with correlation id 65 for topics [Set(json_tweets1)] from broker [BrokerEndPoint(0,localhost,2181)] failed (kafka.client.ClientUtils$) 
java.io.EOFException: Received -1 when reading from channel, socket has likely been closed. 
     at kafka.utils.CoreUtils$.read(CoreUtils.scala:193) 
     at kafka.network.BoundedByteBufferReceive.readFrom(BoundedByteBufferReceive.scala:54) 
     at kafka.network.Receive$class.readCompletely(Transmission.scala:56) 
     at kafka.network.BoundedByteBufferReceive.readCompletely(BoundedByteBufferReceive.scala:29) 
     at kafka.network.BlockingChannel.receive(BlockingChannel.scala:131) 
     at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:77) 
     at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74) 
     at kafka.producer.SyncProducer.send(SyncProducer.scala:115) 
     at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59) 
     at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82) 
     at kafka.producer.BrokerPartitionInfo.getBrokerPartitionInfo(BrokerPartitionInfo.scala:49) 
     at kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$$getPartitionListForTopic(DefaultEventHandler.scala:188) 
     at kafka.producer.async.DefaultEventHandler$$anonfun$partitionAndCollate$1.apply(DefaultEventHandler.scala:152) 
     at kafka.producer.async.DefaultEventHandler$$anonfun$partitionAndCollate$1.apply(DefaultEventHandler.scala:151) 
     at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 
     at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) 
     at kafka.producer.async.DefaultEventHandler.partitionAndCollate(DefaultEventHandler.scala:151) 
     at kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:96) 
     at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:73) 
     at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105) 
     at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:88) 
     at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:68) 
     at scala.collection.immutable.Stream.foreach(Stream.scala:547) 
     at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67) 
     at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45) 

---

Update: Решение

---

1. Пошел в Ambari Services и сменил каталог журналов Kafka на /tmp/kafka-logs.

2. Изменен исходный сценарий, чтобы включить правильный порт и имя хоста.

   python raw_tweets_stream.py | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list sandbox.hortonworks.com:6667 --topic raw_json_tweets

3. Проверено, что события посылаются на тему Кафки с помощью консоли потребителя.

   /usr/hdp/2.3.0.0-2557/kafka/bin/kafka-console-consumer.sh -zookeeper sandbox.hortonworks.com:2181 -topic raw_json_tweets -from-beginning

Answer 1

Похоже, что вы направляете --broker-list на 2181 (зоопарка), а вам нужно указать на Кафку брокера по умолчанию которого порт 9092 или 6667 на Ambari.