Использование Spark для обработки набора данных большего размера, чем может уместиться кластер

Question

Я работаю в кластере Spark 2.3 из 5 узлов, каждый из которых имеет 12 ГБ доступной памяти, и пытаюсь работать с набором данных Parquet размером около 130 ГБ, поверх которого я создал секционированную внешнюю таблицу Hive.

Допустим, я хотел бы знать количество записей в наборе данных.Сначала я предполагал, что Spark будет читать раздел данных по разделам, агрегировать текущий раздел, чтобы получить количество записей, передавать результат драйверу, а затем отбрасывать этот раздел, чтобы прочитать следующий.Однако, либо это не так (скорее, Spark сначала пытается прочитать все в памяти), либо я неправильно его кодирую.

Глупый подход, такой как sql("select count(*) from myhivetable"), не работает: задание завершается с ошибкой пространства кучи Java.sql("select * from myhivetable").count() (я думаю, они все равно скомпилированы в один и тот же план выполнения).

Я мог бы потенциально забыть о таблице кустов, использовать API файловой системы, чтобы получить список файлов, составляющих эту таблицу, и сделать записиподсчет файла за файлом, суммирование суммы в результате, например, fileList.foldLeft{ (recCount, file) => file.read(parquet).count + recCount} - однако, а) этот метод может не "масштабироваться" для других возможных вариантов использования, б) я все еще думаю, что должен быть более элегантныйспособ сделать это просто с помощью набора инструментов Spark.Я не вижу этого?

Answer 1

Предполагая, что у вас есть 8 ядер на узел, вы можете попробовать использовать эти параметры с spark-submit или spark-shell:

Total memory - 5 * 12GB = 60GB Total Cores - 5 * 8 = 40

--driver-cores 1
--driver-memory 2G
--num-executors 9
--executor-cores 4
--executor-memory 6G

Если это не работает,Вы можете изменить номера и попробовать еще раз?и, пожалуйста, опубликуйте журнал ошибок, снимки экрана Spark UI.

Я не уверен, имеет ли какое-то значение использование SELECT COUNT(1) FROM table вместо SELECT COUNT(*) FROM table !!

Answer 2

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