Spark - 没有停止使用 Kafka 主题的 Spark Stream答案

【问题标题】：Spark - Is not stopping Spark Stream that consumes a Kafka topicSpark - 没有停止使用 Kafka 主题的 Spark Stream
【发布时间】：2018-09-21 16:36:28
【问题描述】：

我正在尝试为使用来自 kafka 的数据的火花流示例编写测试。我为此使用EmbeddedKafka。

  implicit val config = EmbeddedKafkaConfig(kafkaPort = 12345)

  EmbeddedKafka.start()
  EmbeddedKafka.createCustomTopic(topic)

  println(s"Kafka Running ${EmbeddedKafka.isRunning}")

  val spark = SparkSession.builder.appName("StructuredStreaming").master("local[2]").getOrCreate
  import spark.implicits._

  val df = spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "localhost:12345")
    .option("subscribe", topic)
    .load()

  // pushing data to kafka
  vfes.foreach(e => {
    val json = ...
    EmbeddedKafka.publishStringMessageToKafka(topic, json)
  })

  val query = df.selectExpr("CAST(value AS STRING)")
    .as[String]
    .writeStream.format("console")

  query.start().awaitTermination()
  spark.stop()
  EmbeddedKafka.stop()

当我运行它时，它会继续运行并且不会停止或向控制台打印任何内容。我不知道为什么会这样。我还尝试通过在流上调用stop 之前调用EmbeddedKafka.stop() 来终止kafka。

【问题讨论】：

Kafka 主题根据定义是无穷无尽的。您需要不同的停止标准，例如输入超时。但两年过去了，也许你已经得出这个结论了:)

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

【解决方案1】：

尝试设置超时

 query.start().awaitTermination( 3000)

其中 3000 以毫秒为单位

【讨论】：