не может потреблять сообщение kafka через искровой поток

Question

Я пытаюсь поглотить сообщение от производителя кафки через программу искрообразования.

Вот моя программа

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)) 
    }) 

выше программа работает успешно. Но я не видел, чтобы какое-либо сообщение печаталось.

Answer 1

Установите мастер URL с помощью "local[*]"

val sparkConf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[*]") 

Вы также можете попытаться вызвать сбор() и посмотреть, если вы получаете сообщения.

lines.foreachRDD { rdd => 
     rdd.collect().foreach(println) 
}