spark streaming 1.6+kafka:太多批次处于“排队”状态

qmelpv7a  于 2021-06-08  发布在  Kafka
关注(0)|答案(1)|浏览(413)

我使用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是实现它的正确方法。
现在,在下图中,您可以看到有许多批正在排队,而正在处理的批需要很长时间才能完成。

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

当批处理中没有太多事件时,为什么insert操作有这么多任务?有时,拥有0个事件也会产生数千个任务,这些任务需要永远才能完成。
我用spark处理微博客的方式是错误的吗?
谢谢你的帮助!
一些额外的细节:
Yarn容器的最大容量为2gb。在这个Yarn队列中,容器的最大数量是10个。当我查看执行这个spark应用程序的队列的详细信息时,容器的数量非常大,大约有15k个挂起的容器。

wd2eg0qa

wd2eg0qa1#

我终于明白了。显然,spark流不能处理空事件,因此在代码的foreachrdd部分中,我添加了以下内容:

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

这样我们就跳过了空的微批次。isempty()方法无效。
希望这能帮助别人!;)

相关问题