Ошибка контрольной суммы блоков данных при записи в файл

Question

Я выполняю работу в 9 узлах.

Все они собираются записать некоторую информацию в файлы, делая простые записи, как показано ниже:

dfLogging.coalesce(1).write.format('delta').mode('append').save('/dbfs/' + loggingLocation)

Однако я получаю это исключение:

py4j.protocol.Py4JJavaError: Произошла ошибка при вызове o106.save.: java.util.concurrent.ExecutionException: org.apache.spark.SparkException: задание прервано из-за сбоя этапа: сбой задачи 1 на этапе 14.0 1 раз, последний сбой: потерянное задание 1.0 на этапе 14.0 (TID 259, localhost, executor драйвер): org.apache.hadoop.fs.ChecksumException: ошибка контрольной суммы: файл:/dbfs/delta/Logging/_delta_log/00000000000000000063.json в 0 exp: 1179219224 получил: -1020415797

Мне кажется, что из-за параллелизма спарк как-то не работает, и он генерирует ошибки контрольной суммы.

Есть ли какой-либо известный сценарий, который может быть причиной этого?

Answer 1

Таким образом, происходит несколько вещей, и это должно объяснить, почему объединение может не работать.

1. Что объединяет, так это то, что он по существу объединяет разделы на каждом работнике. Например, если у вас три работника, вы можете выполнить coalesce(3), которая объединит разделы на каждом работнике.

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

Почему объединение не будет работать? Искра ограничивает перетасовку во время объединения. Таким образом, вы не можете выполнить полное перемешивание (для разных работников), когда используете объединение, тогда как вы можете выполнить полное перемешивание, когда используете перераспределение, хотя это дорогостоящая операция.

Вот код, который будет работать:

dfLogging.repartition(1).write.format('delta').mode('append').save('/dbfs/' + loggingLocation)