Spark Driver не хватает памяти из-за связи с исполнителями ⇐ JAVA

[Anonymous]

Мой драйвер Spark не работает с OOM:

Код: Выделить всё

java.lang.OutOfMemoryError: Java heap space
  at java.util.Arrays.copyOf(Arrays.java:3332)
  at java.lang.AbstractStringBuilder.ensureCapacityInternal(AbstractStringBuilder.java:124)
  at java.lang.AbstractStringBuilder.append(AbstractStringBuilder.java:448)
  at java.lang.StringBuilder.append(StringBuilder.java:141)
  at java.lang.StringBuilder.append(StringBuilder.java:136)
  at java.util.AbstractCollection.toString(AbstractCollection.java:462)
  at java.util.Collections$UnmodifiableCollection.toString(Collections.java:1037)
  at org.apache.spark.util.JsonProtocol$.accumValueToJson(JsonProtocol.scala:469)
  at org.apache.spark.util.JsonProtocol$.$anonfun$accumulableInfoToJson$7(JsonProtocol.scala:428)
  at org.apache.spark.util.JsonProtocol$$$Lambda$4808/1475607995.apply(Unknown Source)
  at scala.Option.map(Option.scala:230)
  at org.apache.spark.util.JsonProtocol$.accumulableInfoToJson(JsonProtocol.scala:428)
  at org.apache.spark.util.JsonProtocol$.$anonfun$accumulablesToJson$4(JsonProtocol.scala:420)
  at org.apache.spark.util.JsonProtocol$$$Lambda$4204/1507248734.apply(Unknown Source)
  at scala.collection.immutable.List.map(List.scala:293)
  at org.apache.spark.util.JsonProtocol$.accumulablesToJson(JsonProtocol.scala:420)
  at org.apache.spark.util.JsonProtocol$.taskInfoToJson(JsonProtocol.scala:412)
  at org.apache.spark.util.JsonProtocol$.taskEndToJson(JsonProtocol.scala:171)
  at org.apache.spark.util.JsonProtocol$.sparkEventToJson(JsonProtocol.scala:81)
  at org.apache.spark.deploy.history.rpc.app.RpcAppEvent$.encodeEvent(RpcAppEvent.scala:77)
  at org.apache.spark.deploy.history.rpc.app.RpcAppEvent$.apply(RpcAppEvent.scala:62)
  at org.apache.spark.deploy.history.rpc.app.RpcAppEventQueue.enqueue(RpcAppEventQueue.scala:86)
  at org.apache.spark.deploy.history.rpc.app.RpcAppListener.addToEventQueue(RpcAppListener.scala:75)
  at org.apache.spark.deploy.history.rpc.app.RpcAppListener.onEvent(RpcAppListener.scala:83)
  at org.apache.spark.deploy.history.rpc.util.SingleHandlerListener.onTaskEnd(SingleHandlerListener.scala:48)
  at org.apache.spark.scheduler.SparkListenerBus.doPostEvent(SparkListenerBus.scala:45)
  at org.apache.spark.scheduler.SparkListenerBus.doPostEvent$(SparkListenerBus.scala:28)
  at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37)
  at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37)
  at org.apache.spark.util.ListenerBus.postToAll(ListenerBus.scala:120)
  at org.apache.spark.util.ListenerBus.postToAll$(ListenerBus.scala:104)
  at org.apache.spark.scheduler.AsyncEventQueue.super$postToAll(AsyncEventQueue.scala:127)

Конфигурация Spark:

Код: Выделить всё

spark.dynamicAllocation.maxExecutors=9
spark.executor.cores=2
spark.dynamicAllocation.minExecutors=1
spark.dynamicAllocation.initialExecutors=1
spark.executor.instances=8
spark.shuffle.service.client.class=org.apache.spark.network.shuffle.ExternalShuffleClient
spark.driver.maxResultSize=1g
spark.sql.files.maxPartitionBytes=1073741824
spark.sql.join.preferSortMergeJoin=true
spark.dynamicAllocation.executorIdleTimeout=300s
spark.executor.extraJavaOptions=-Detwlogger.component=sparkexecutor -DlogFilter.filename=SparkLogFilters.xml -DpatternGroup.filename=SparkPatternGroups.xml -Dlog4jspark.root.logger=INFO,console,RFA,ETW,Anonymizer -Dlog4jspark.log.dir=/var/log/sparkapp/${user.name} -Dlog4jspark.log.file=sparkexecutor.log -Dlog4j2.configurationFile=file:/usr/hdp/current/spark3-client/conf/executor-log4j2.properties -Djavax.xml.parsers.SAXParserFactory=com.sun.org.apache.xerces.internal.jaxp.SAXParserFactoryImpl -XX:+UseParallelGC -XX:+UseParallelOldGC
spark.driver.cores=1
spark.driver.memory=**2g**
spark.executor.memory=9g
spark.driver.memoryOverhead=**384**
spark.executor.memoryOverhead=921

Всего имеется 9 исполнителей и 919 задач.
Код выполняется внутри:

Код: Выделить всё

df.rdd.foreachPartition

С некоторыми аккумуляторами целых и строк (строковые аккумуляторы контролируются, чтобы не расти бесконечно)
После завершения каждой задачи я могу видеть этот журнал (для каждой задачи это примерно ~60000-62000, никаких выбросов):

Код: Выделить всё

INFO Executor [Executor task launch worker for task 258.0 in stage 59.0 (TID 4666)]: Finished task 258.0 in stage 59.0 (TID 4666).  61394 bytes result sent to driver

И Spark дает сбой с OOM после выполнения примерно 270/919 задач.
Насколько я понимаю, он терпит неудачу, когда Исполнители сообщают свое состояние Драйверу и Драйверу накапливает эту информацию, выполняя все функции JSON.
Однако 270 * 60 КБ ~ 15 МБ и не следует убивать Driver с помощью OOM, если только он не копируется несколько раз внутри системы.
Самое простое решение — добавить память в драйвер, но мне интересно, как я могу исследовать этот OOM дальше, чтобы доказать свою теорию о том, что OOM произошел из-за связи между Исполнительом -> Драйвером
Спасибо!

Подробнее здесь: https://stackoverflow.com/questions/792 ... -executors