我试图消费Kafka生产者通过Spark流节目的信息。
这是我的程序
val Array(zkQuorum, group, topics, numThreads) = args
val sparkConf = new SparkConf().setAppName("KafkaWordCount").setMaster("local")
val ssc = new StreamingContext(sparkConf, Seconds(5))
val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)
// lines.print()
lines.foreachRDD(rdd=>{
rdd.foreach(message=>
println(message))
})以上程序正在成功运行。但我看不到任何信息被打印出来。
1条答案
按热度按时间w6lpcovy1#
使用设置主url
"local[*]"```val sparkConf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[*]")
lines.foreachRDD { rdd =>
rdd.collect().foreach(println)
}