无法通过火花流消费卡夫卡消息

【问题标题】：unable to consume kafka message through spark stream无法通过火花流消费卡夫卡消息
【发布时间】：2017-06-29 19:16:07
【问题描述】：

我正在尝试通过火花流程序消耗来自 kafka 生产者的消息。

这是我的程序

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

以上程序运行成功。但我看不到任何消息被打印出来。

【问题讨论】：

标签： apache-spark apache-kafka spark-streaming kafka-consumer-api

【解决方案1】：

使用"local[*]"设置您的主网址

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

你也可以尝试调用collect()，看看是否收到消息。

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

【讨论】：

我不想使用 collect 。它效率不高，因为我将获得 10 k 到 50 k 记录。
我测试了您为打印而编写的循环，这对我来说效果很好。如果您收到消息，我只是想让您使用 collect() 检查。检查您的主题名称。如果可以看到消息，您是否尝试过使用 kafka-console-consumer？
我可以在 kafka 消费者和使用 collect() 中看到消息。你测试了什么我不明白？