17

我正在使用 java 1.8.0_45 和 Kafka 直接流在 Yarn（Apache 发行版 2.6.0）上运行 spark 流式传输 1.4.0。我也在使用带有 scala 2.11 支持的 spark。

我看到的问题是驱动程序和执行程序容器都在逐渐增加物理内存使用量，直到纱线容器杀死它。我在我的驱动程序中配置了多达 192M 堆和 384 个堆外空间，但它最终用完了

堆内存在常规 GC 周期中似乎很好。在任何此类运行中都没有遇到过 OutOffMemory

事实上，我并没有在 kafka 队列上产生任何流量，但仍然会发生这种情况。这是我正在使用的代码

object SimpleSparkStreaming extends App {

val conf = new SparkConf()
val ssc = new StreamingContext(conf,Seconds(conf.getLong("spark.batch.window.size",1L)));
ssc.checkpoint("checkpoint")
val topics = Set(conf.get("spark.kafka.topic.name")); 
    val kafkaParams = Map[String, String]("metadata.broker.list" -> conf.get("spark.kafka.broker.list"))
            val kafkaStream = KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc, kafkaParams, topics)
            kafkaStream.foreachRDD(rdd => {
                rdd.foreach(x => {
                    println(x._2)
                })

            })
    kafkaStream.print()
            ssc.start() 

            ssc.awaitTermination()

}

我在 CentOS 7 上运行它。用于 spark 提交的命令如下

./bin/spark-submit --class com.rasa.cloud.prototype.spark.SimpleSparkStreaming \
--conf spark.yarn.executor.memoryOverhead=256 \
--conf spark.yarn.driver.memoryOverhead=384 \
--conf spark.kafka.topic.name=test \
--conf spark.kafka.broker.list=172.31.45.218:9092 \
--conf spark.batch.window.size=1 \
--conf spark.app.name="Simple Spark Kafka application" \
--master yarn-cluster \
--num-executors 1 \
--driver-memory 192m \
--executor-memory 128m \
--executor-cores 1 \
/home/centos/spark-poc/target/lib/spark-streaming-prototype-0.0.1-SNAPSHOT.jar

任何帮助是极大的赞赏

问候，

阿普尔瓦

4

3 回答 3

1

尝试增加执行器核心。在您的示例中，唯一的核心专用于使用流数据，没有核心在传入数据中进行处理。

于 2016-03-17T08:36:56.780 回答
0

这不是 Kafka 的答案，这将与 Spark 隔离，以及它的编目系统在一致的持久性和大型操作方面是多么糟糕。如果您一直在写入一个 perisitence 层（即在一个循环中，在一个大型操作之后重新持久化一个 DF，然后再次运行）或运行一个大型查询（即 inputDF.distinct.count）；Spark 作业将开始将一些数据放入内存中，并且无法有效地删除过时的对象。

这意味着超时一个能够快速运行一次的对象，将稳步减慢，直到没有可用的内存。对于家里的每个人，启动一个在环境中加载了大型 DataFrame 的 AWS EMR，运行以下查询：

var iterator = 1
val endState = 15
var currentCount = 0
while (iterator <= endState) {
  currentCount = inputDF.distinct.count
  print("The number of unique records are : " + currentCount)
  iterator = iterator + 1
}

在作业运行时观察 Spark UI 的内存管理，如果 DF 对会话来说足够大，您将开始注意到每次后续运行的运行时间下降，主要是块变得陈旧但 Spark 无法识别何时清理这些块。

我找到解决此问题的最佳方法是在本地编写我的 DF，清除持久层并重新加载数据。这是解决问题的“大锤”方法，但对于我的业务案例来说，它是很容易实现的解决方案，它导致我们的大型表的运行时间增加了 90%（需要 540 分钟到大约 40 分钟，内存更少）。

我目前使用的代码是：

val interimDF = inputDF.action
val tempDF = interimDF.write.format(...).option("...","...").save("...")
spark.catalog.clearCache
val interimDF = spark.read..format(...).option("...","...").save("...").persist
interimDF.count

如果您不在子子流程中取消保留 DF，那么这里是一个派生词：

val interimDF = inputDF.action
val tempDF = interimDF.write.format(...).option("...","...").save("...")
for ((k,v) <- sc.getPersistentRDDs) {
  v.unpersist()
}
val interimDF = spark.read..format(...).option("...","...").save("...").persist
interimDF.count
于 2019-08-12T17:47:05.633 回答
0

这可能是内存泄漏...您是否尝试过 conf.set("spark.executor.extraJavaOptions","-XX:+UseG1GC") ？

于 2016-07-20T08:40:32.600 回答