Spark Streaming 1.6 + Kafka：处于“排队”状态的批次过多答案

【问题标题】：Spark Streaming 1.6 + Kafka: Too many batches in "queued" statusSpark Streaming 1.6 + Kafka：处于“排队”状态的批次过多
【发布时间】：2018-12-26 14:03:59
【问题描述】：

我正在使用 spark 流来使用来自 Kafka 主题的消息，该主题有 10 个分区。我正在使用直接方法从 kafka 消费，代码如下：

def createStreamingContext(conf: Conf): StreamingContext = {
    val dateFormat = conf.dateFormat.apply
    val hiveTable = conf.tableName.apply

    val sparkConf = new SparkConf()

    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    sparkConf.set("spark.driver.allowMultipleContexts", "true")

    val sc = SparkContextBuilder.build(Some(sparkConf))
    val ssc = new StreamingContext(sc, Seconds(conf.batchInterval.apply))

    val kafkaParams = Map[String, String](
      "bootstrap.servers" -> conf.kafkaBrokers.apply,
      "key.deserializer" -> classOf[StringDeserializer].getName,
      "value.deserializer" -> classOf[StringDeserializer].getName,
      "auto.offset.reset" -> "smallest",
      "enable.auto.commit" -> "false"
    )

    val directKafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
      ssc,
      kafkaParams,
      conf.topics.apply().split(",").toSet[String]
    )

    val windowedKafkaStream = directKafkaStream.window(Seconds(conf.windowDuration.apply))
    ssc.checkpoint(conf.sparkCheckpointDir.apply)

    val eirRDD: DStream[Row] = windowedKafkaStream.map { kv =>
      val fields: Array[String] = kv._2.split(",")
      createDomainObject(fields, dateFormat)
    }

    eirRDD.foreachRDD { rdd =>
      val schema = SchemaBuilder.build()
      val sqlContext: HiveContext = HiveSQLContext.getInstance(Some(rdd.context))
      val eirDF: DataFrame = sqlContext.createDataFrame(rdd, schema)

      eirDF
        .select(schema.map(c => col(c.name)): _*)
        .write
        .mode(SaveMode.Append)
        .partitionBy("year", "month", "day")
        .insertInto(hiveTable)
    }
    ssc
  }

从代码中可以看出，我使用 window 来实现这一点（如果我错了请纠正我）：由于有一个动作可以插入到 hive 表中，所以我想要避免过于频繁地写入 HDFS，所以我想要的是在内存中保存足够的数据，然后才写入文件系统。我认为使用 window 是实现它的正确方法。

现在，在下图中，您可以看到有很多批次正在排队，正在处理的批次需要永远完成。

我还提供了正在处理的单个批次的详细信息：

当批处理中的事件不多时，为什么插入操作有这么多任务？有时，事件为 0 也会产生数千个任务，这些任务需要很长时间才能完成。

我使用 Spark 处理微批处理的方式有误吗？

感谢您的帮助！

一些额外的细节：

纱线容器的最大容量为 2gb。在这个 Yarn 队列中，容器的最大数量是 10。当我查看正在执行此 spark 应用程序的队列的详细信息时，容器的数量非常大，大约 15k 待处理的容器。

【问题讨论】：

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

【解决方案1】：

好吧，我终于想通了。显然 Spark Streaming 无法处理空事件，因此在代码的 foreachRDD 部分中，我添加了以下内容：

eirRDD.foreachRDD { rdd =>
      if (rdd.take(1).length != 0) {
        //do action
      }
}

这样我们就可以跳过空的微批次。 isEmpty() 方法不起作用。

希望这对其他人有帮助！ ;)

【讨论】：