如何在 Spark Streaming 中从头开始读取 Kafka 主题的记录？答案

【问题标题】：How to read records from Kafka topic from beginning in Spark Streaming?如何在 Spark Streaming 中从头开始读取 Kafka 主题的记录？
【发布时间】：2017-04-11 02:18:30
【问题描述】：

我正在尝试使用 Spark Streaming 从 Kafka 主题中读取记录。

这是我的代码：

object KafkaConsumer {

  import ApplicationContext._

  def main(args: Array[String]) = {

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> s"${UUID.randomUUID().toString}",
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topics = Array("pressure")
    val stream = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )
    stream.print()
    stream.map(record => (record.key, record.value)).count().print()
    streamingContext.start()
  }
}

当我运行它时它什么也不显示。

为了检查pressure 主题中是否确实存在数据，我使用了命令行方法，它确实显示了记录：

bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic pressure \
  --from-beginning

输出：

TimeStamp:07/13/16 15:20:45:226769,{'Pressure':'834'}
TimeStamp:07/13/16 15:20:45:266287,{'Pressure':'855'}
TimeStamp:07/13/16 15:20:45:305694,{'Pressure':'837'}

怎么了？

【问题讨论】：

标签： scala apache-spark apache-kafka spark-streaming

【解决方案1】：

你错过了streamingContext.awaitTermination()。

【讨论】：

【解决方案2】：

您需要启动streamingContext，最后执行streamingContext.awaitTermination()。

【讨论】：