Spark saveAsTextFile с BZip2Codec вызывает утечку памяти

Question

Я использую Spark 2.1.0 на EMR 5.5.0 (Java 1.8.0_121) с YARN в качестве менеджера ресурсов.

Я столкнулся с проблемой с искрой, когда через несколько часов Yarn убивает исполнителя после того, как он израсходовал всю физическую память сконфигурированного контейнера. Я обнаружил, что источник этой проблемы связан с Bzip2Codec, в частности: rdd.saveAsTextFile(path,classOf[BZip2Codec])

Если я удаляю сжатие или меняю его на другой кодек сжатия, проблема не возникает.

моя искра-подать:

spark-submit \
  --deploy-mode cluster \
  --master yarn \
  --executor-memory 1g \
  --num-executors 1 \
  --conf spark.yarn.executor.memoryOverhead=384 \
  --conf spark.yarn.submit.waitAppCompletion=false \
  --conf spark.streaming.backpressure.enabled=true \
  --conf spark.streaming.kafka.maxRatePerPartition=300 \
  --conf spark.streaming.dynamicAllocation.enabled=false \
  --conf spark.dynamicAllocation.enabled=false \
  --class com.kafkaStreaming.main.Main \
  /path/to/jar.jar

Сам по себе код очень прост, он получает поток от Kafka и выполняет foreachRDD, чтобы сохранить его на S3:

import kafka.serializer.StringDecoder
import org.apache.hadoop.io.compress.BZip2Codec
import org.joda.time.format.DateTimeFormat
import org.joda.time.{DateTime, DateTimeZone}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext, Time}

object Main {
  def main(args: Array[String]) {

    val outputDirectory = "s3://bucketName/sparkOutput/"
    val topic = "TopicName"
    val topicsSet = topic.split(",").toSet
    val kafkaBroker = 10.0.0.10:9092
    val kafkaParams = Map[String,String](("metadata.broker.list" -> kafkaBroker))
    val batchInterval = 5

    val sparkConf = new SparkConf().setAppName("sparkStreaming_tests")
    val ssc = new StreamingContext(sparkConf, Seconds(batchInterval))

    val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
      ssc, kafkaParams, topicsSet)

    stream.map(_._2).foreachRDD {( rdd,time:Time) =>
      val batchTime=new DateTime(time.milliseconds)
      rdd.saveAsTextFile(
        outputDirectory +
          "/" + "part_date=" + batchTime.toString(DateTimeFormat.forPattern("yyyy-MM-dd")) +
          "/" + "part_hour=" + batchTime.toString(DateTimeFormat.forPattern("HH")) +
          "/" + batchTime.toString(DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss"))
        , classOf[BZip2Codec])
    }

    ssc.start()
    ssc.awaitTermination()
  }
}

В поисках Google я наткнулся на эту Jira: https://issues.apache.org/jira/browse/HADOOP-12619

Кто-нибудь сталкивался с этой проблемой? Есть ли известные обходные пути для этой проблемы?