Испускайте утечку памяти кучи на пряжу с прямым потоком Кафки

Я запускаю искровой поток 1.4.0 на пряжу (дистрибутив Apache 2.6.0) с java 1.8.0_45, а также прямой поток Kafka. Я также использую искру с поддержкой scala 2.11.

Проблема, которую я вижу, заключается в том, что оба контейнера драйвера и исполнителя постепенно увеличивают использование физической памяти до момента, когда контейнер с пряжей убивает его. Я сконфигурировал до 192M Heap и 384 от места кучи в моем драйвере, но в итоге он исчерпал его

Память Heap выглядит нормально с обычными циклами GC. Не существует OutOffMemory, встречающегося когда-либо в любых таких прогонах

Infact Я не генерирую никакого трафика в очередях 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. Команда, используемая для получения очереди, выглядит следующим образом

./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

Любая помощь приветствуется

Привет,

Апурва

Ответ 1

Попробуйте увеличить количество ядер исполнителей. В вашем примере единственное ядро предназначено для потребления потоковых данных, не оставляя никаких сердечников для обработки входящих данных.

Ответ 2

Это может быть утечка памяти... Вы пытаетесь с conf.set( "spark.executor.extraJavaOptions", "- XX: + UseG1GC" )?

Ответ 3

Это не ответ кафки, он будет изолирован от Spark и от того, насколько плоха его система каталогизации, когда речь идет о постоянстве и больших операциях. Если вы последовательно записываете на уровень постоянства (то есть в цикле повторно сохраняете DF после большой операции, затем выполняете снова) или выполняете большой запрос (то есть inputDF.distinct.count); задание Spark начнет помещать некоторые данные в память и не будет эффективно удалять устаревшие объекты.

Это означает, что сверхурочно объект, который был в состоянии быстро запустить один раз, будет постоянно замедляться, пока не останется свободной памяти. Для всех, кто работает дома, разверните AWS EMR с большим DataFrame, загруженным в среду, выполните следующий запрос:

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