FetchFailedException или MetadataFetchFailedException при обработке большого набора данных

Ответ 1

Эта ошибка почти гарантированно вызвана проблемами памяти ваших исполнителей. Я могу придумать пару способов решения этих проблем.

1) Вы можете попробовать запустить с большим количеством разделов (сделайте repartition на dataframe). Проблемы с памятью обычно возникают, когда один или несколько разделов содержат больше данных, чем вписываются в память.

2) Я замечаю, что вы явно не установили spark.yarn.executor.memoryOverhead, поэтому по умолчанию будет max(386, 0.10* executorMemory), который в вашем случае будет 400 МБ. Это звучит низко для меня. Я попытался бы увеличить его, чтобы сказать 1 ГБ (учтите, что если вы увеличиваете memoryOverhead до 1 ГБ, вам нужно опустить --executor-memory до 3 ГБ)

3) Посмотрите в файлах журнала на сбоях узлов. Вы хотите найти текст "Killing container". Если вы видите, что текст "выходит за пределы физической памяти", увеличение памяти. На мой взгляд, проблема с памятью будет решена.

Ответ 2

У меня также были хорошие результаты, увеличив время ожидания Spark spark.network.timeout до большего значения, например 800. По умолчанию 120 секунд заставит многих ваших исполнителей тайм-аут при большой нагрузке.

Ответ 3

Хорошо, это старая нить, и в Stackoverflow есть довольно много ответов, но я потерял пару дней до этой ошибки, и я думаю, что совместная история может помочь.

На самом деле это может быть несколько способов. Как упоминал Гленни, это скорее всего проблема памяти, поэтому убедитесь, что у вас достаточно памяти для всего. Есть память контейнера, память AM, память памяти, сокращение памяти и т.д. Конфигурации, которые нужно учитывать. Чтение this может помочь вам найти правильные конфигурации. Вы должны сами выбирать номера, но вот несколько свойств, которые я установил.

пряжи site.xml

<property>
    <name>yarn.nodemanager.resource.memory-mb</name>
    <value>32768</value>
</property>

<property>
    <name>yarn.app.mapreduce.am.resource.mb</name>
    <value>4096</value>
</property>

<property>
    <name>yarn.scheduler.minimum-allocation-mb</name>
    <value>4096</value>
</property>

mapred-site.xml

<property>
    <name>mapreduce.map.memory.mb</name>
    <value>4096</value>
</property>

<property>
    <name>mapreduce.reduce.memory.mb</name>
    <value>4096</value>
</property>

Они могут исправить некоторые другие ошибки, с которыми вы могли столкнуться, например, сбой при работе с PySpark при запуске. Но в моем случае, хотя некоторые ошибки исчезли (например, ошибки MetadataFetchFailed), проблема не устранена. Точная ошибка:

org.apache.spark.shuffle.FetchFailedException: не удалось подключиться к DB-ЕТ-С/х.х.х.м: 34085

После игры со всеми возможными свойствами YARN и Spark от тайм-аутов Spark к сервису Shuffle YARN я понял, что в журналах ошибок неудавшийся контейнер ищет x.x.x.x, локальный (внутренний) IP при запуске netstat -tulpn | grep <PORT NUM> вернулся yyyy: 34085, в котором yyyy является внешним IP-адресом. Это не проблема памяти, это была проблема с настройкой сети.

Служба Spark была привязана только к внешнему интерфейсу, поскольку имя хоста было связано с внешним IP-адресом в /etc/hosts. После обновления файла /etc/hosts проблема была исправлена.

Нижняя линия: Очевидно, ошибка говорит о том, что какой-то контейнер не может достичь другого. Обычно это связано с неудачными контейнерами из-за проблем с памятью, но также может быть сетевой проблемой, поэтому следите за ними, особенно если у вас есть несколько интерфейсов на ваших узлах.

Ответ 4

В дополнение к проблемам с памятью и сетевыми конфигурациями, описанным выше, стоит отметить, что для больших таблиц (например, несколько ТБ здесь) может возникнуть проблема org.apache.spark.shuffle.FetchFailedException из-за тайм-аута, получающего перетасованные разделы. Чтобы устранить эту проблему, вы можете установить следующее:

SET spark.reducer.maxReqsInFlight=1;  -- Only pull one file at a time to use full network bandwidth.
SET spark.shuffle.io.retryWait=60s;  -- Increase the time to wait while retrieving shuffle partitions before retrying. Longer times are necessary for larger files.
SET spark.shuffle.io.maxRetries=10;