Есть ли обходной путь для «org. apache .spark.SparkException: сбой сериализации Kryo: переполнение буфера. Доступно: 0, обязательно: n» ошибка?

Question

Я выполняю задание pyspark через spark-submit, в котором две большие паркетные таблицы сначала соединяются вместе перед применением некоторой дополнительной логики c. После отправки все задачи завершаются неудачно, и в журналах указывается следующее:

Caused by: org.apache.spark.SparkException: Kryo serialization failed: Buffer overflow. Available: 0, required: n*. To avoid this, increase spark.kryoserializer.buffer.max value

* n - это просто переменная, представляющая, сколько памяти требуется. Это значение зависит от того, насколько я установил spark.kryoserializer.buffer.max в значение.

Я увеличил значение этого параметра до 2047 МБ и больше не вижу ошибку, и задание отображается как «Выполнено» в трекере заданий ApplicationMaster. но задание по-прежнему отображается как Запуск на консоли, а также в Yarn.

Контекст для запускаемого задания:

Для контекста, вот что я делаю. Я беру очень большой список URL-адресов и объединяю их в один и тот же точный список с критериями совпадения, что вторая строка URL-адреса содержится в первой строке URL-адреса. Например: www.google.com будет соответствовать www.google.com и www.google.com/1 и www.google.com/1/2 URL_Hierarachy = URL_1.join (URL_2, URL_1.url1 == URL_2.url2.substr (URL_1.numone, URL_1.nchar_url1), «влево» ) где numone и nchar_url1 используются для определения длины первого URL. Это даст гораздо больше записей, чем начальные 93 млн.

Answer 1

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

И в то же время spark2 + использует kryoserializer в качестве сериализатора по умолчанию. Вы получите эту ошибку.

Я бы дважды проверил код вашего объединения, раздела фрейма данных, посмотрел бы, достаточно ли большого фрейма данных spark имеет раздел / или есть ли асимметрия в данных.

для увеличения параллелизма происходит при объединении

spark-submit --conf spark.sql.shuffle.partitions=2000

для увеличения раздела заданного фрейма данных

df.repartition(2000)