Kafka producer to read data files

apache-kafka kafka-producer-api
Sumit picture Sumit · Feb 13, 2016 · Viewed 20.3k times · Source

I am trying to load a data file in loop(to check stats) instead of standard input in Kafka. After downloading Kafka, I performed the following steps:

Started zookeeper:

bin/zookeeper-server-start.sh config/zookeeper.properties

Started Server:

bin/kafka-server-start.sh config/server.properties

Created a topic named "test":

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

Ran the Producer:

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test 
Test1
Test2

Listened by the Consumer:

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
Test1
Test2

Instead of Standard input, I want to pass a data file to the Producer which can be seen directly by the Consumer. Or is there any kafka producer instead of console consumer using which I can read data files. Any help would really be appreciated. Thanks!

Answer

Shawn Guo picture Shawn Guo · Feb 13, 2016

You can read data file via cat and pipeline it to kafka-console-producer.sh. 

cat ${datafile} | ${kafka_home}/bin/kafka-console-producer.sh --broker-list ${brokerlist} --topic test

Related questions

How to check whether Kafka Server is running?

I want to ensure whether kafka server is running or not before starting production and consumption jobs. It is in windows environment and here's my kafka server's code in eclipse... Properties kafka = new Properties(); kafka.setProperty("broker.id", "1"); kafka.setProperty("…

Read More
Understanding Kafka Topics and Partitions

I am starting to learn Kafka for enterprise solution purposes. During my readings, some questions came to my mind: When a producer is producing a message - it will specify the topic it wants to send the message to, is …

Read More
Is key required as part of sending messages to Kafka?

KeyedMessage<String, byte[]> keyedMessage = new KeyedMessage<String, byte[]>(request.getRequestTopicName(), SerializationUtils.serialize(message)); producer.send(keyedMessage); Currently, I am sending messages without any key as part of keyed messages, will it still work with delete.retention.…

Read More