У меня довольно простая искровая работа, которая объединяет несколько наборов данных. Код примерно следующий:
List<Dataset<Row>> inputs; // about 20 input datasets
Dataset<Row> output = inputs.stream().reduce(Dataset::union).map(ds -> ds.repartition(600))
writeToHDFS(output)
Когда я запускаю это и смотрю на задание в YARN, я вижу, что все это выполняется в драйвере. (здесь нет никаких фильтров или вычислений).
Однако, когда я начинаю работу, я вижу, что она начинает что-то делать, пока не выполнит полный сбор мусора и не погибнет от YARN через 2 минуты после ее запуска.
Все входные наборы данных довольно малы (от 1 до 20 ГБ), и я уже запускаю их со следующей конфигурацией искры:
spark.driver.cores: 8
spark.driver.memory: 28g
Ребята, у вас есть еще идеи, как я могу добиться успеха профсоюза? Может ли это быть как-то распространено, чтобы не было OOM драйвера?
Я использовал функцию writeToHDFS (пользовательский код) с наборами данных, которые имеют большие ТБ и не имеют проблем, поэтому я сомневаюсь, что это часть для хранения. Это скорее сериализация, которая обычно, вероятно, читает результаты от исполнителей, но в случае union пытается собрать весь фрагмент в драйвере.