Spark Structured Streaming OutOfMemoryError, вызванная тысячами экземпляров KafkaMbean

Question

Исполнитель Spark Structured Streaming завершается с ошибкой OutOfMemoryError

Проверка выделения кучи с помощью VirtualVM указывает на то, что использование памяти JMX Mbean Server линейно растет со временем.

После дальнейшего исследования кажется, что JMX Mbeanзаполнены тысячами экземпляров объектов KafkaMbean с метриками для потребителя - (\ d +), которые исчисляются тысячами (равным количеству задач, созданных на исполнителе).

Запуск потребителя Kafka с журналами DEBUG на исполнителе показываетчто исполнитель добавляет тысячи датчиков метрик и часто вообще не удаляет их или только удаляет некоторые

Я использую HDP Spark 2.3.0.2.6.5.0-292 с HDP Kafka 1.0.0.2.6.5.0-292.

Вот как я инициализирую структурированную потоковую передачу:

sparkSession
  .readStream
  .format("kafka")
  .options(Map("kafka.bootstrap.servers" -> KAFKA_BROKERS,
               "subscribePattern"        -> INPUT_TOPIC,
               "startingOffsets"         -> "earliest",
               "failOnDataLoss"          -> "false"))
  .mapPartitions(processData)
  .writeStream
  .format("kafka")
  .options(Map("kafka.bootstrap.servers" -> KAFKA_BROKERS,
               "checkpointLocation" -> CHECKPOINT_LOCATION))
  .queryName("Process Data")
  .outputMode("update")
  .trigger(Trigger.ProcessingTime(1000))
  .load()
  .start()
  .awaitTermination()

Я ожидал, что Spark / Kafka должным образом очистит MBeans при завершении задачи, но, похоже, это не так.

Answer 1

Ваша версия HDP, вероятно, использует spark 2.3.1, в которой есть известная ошибка чтения данных из Kafka (эта проблема возникает, когда вы читаете из темы, в которой нет новых данных в каждой микропакете):

https://issues.apache.org/jira/browse/SPARK-24987

https://issues.apache.org/jira/browse/SPARK-25106

Эта ошибка возникла в результате изменения, внесенного в версию 2.3.1 (в версии 2.3.0 ее нет), поэтому выВы можете обновить версию Spark или получить патч для своей версии HDP.