1,将SPARK_HOME中的DirectKafkaWordCount程序复制到idea中.
2,由于在KafkaWordCount中已引入相关jar包,此步可略过
3,配置configurations参数
关于program arguments参数,程序中定义如下:
/**
* Consumes messages from one or more topics in Kafka and does wordcount.
* Usage: DirectKafkaWordCount <brokers> <topics>
* <brokers> is a list of one or more Kafka brokers
* <topics> is a list of one or more kafka topics to consume from
*
* Example:
* $ bin/run-example streaming.DirectKafkaWordCount broker1-host:port,broker2-host:port
* topic1,topic2
*/
第一个参数为一系列的brokers,可以设置多个;第二个参数为一系列topic,也可以设置为多个.
本次运行时只采用localhost:9092和test
4,将之前的KafkaWordCountProducer运行起来,运行DirectKafkaWordCount可以看到如下结果:
也可不运行KafkaWordCountProducer,而是另外起一个console服务,在窗口中运行kafka-console-producer.sh --broker-list localhost:9092 --sync --topic test
