Apache Spark не обрабатывает большое семейство столбцов Cassandra

Question

Я пытаюсь использовать Apache Spark для обработки моих больших (~ 230 тыс. Записей) набора данных cassandra, но я постоянно сталкиваюсь с различными видами ошибок. Однако я могу успешно запускать приложения при работе в наборе данных ~ 200 записей. У меня есть искровая установка из 3 узлов с 1 мастером и 2 работниками, а у 2 рабочих также есть кластер cassandra, с данными, индексированными с коэффициентом репликации 2. Мои 2 искрогасителя показывают 2,4 и 2,8 ГБ памяти на веб-интерфейсе и Я установил spark.executor.memory на 2409 при запуске приложения, чтобы получить суммарную память 4,7 ГБ. Вот мой WebUI Главная

среда страница одной из задач

На данном этапе, я просто пытаюсь обрабатывать данные, хранящиеся в Кассандре с помощью искры. Вот основной код, я использую, чтобы сделать это в Java

SparkConf conf = new SparkConf(true) 
     .set("spark.cassandra.connection.host", CASSANDRA_HOST) 
     .setJars(jars); 

SparkContext sc = new SparkContext(HOST, APP_NAME, conf); 
SparkContextJavaFunctions context = javaFunctions(sc); 

CassandraJavaRDD<CassandraRow> rdd = context.cassandraTable(CASSANDRA_KEYSPACE, CASSANDRA_COLUMN_FAMILY); 

System.out.println(rdd.count()); 

Для успешного запуска, на небольшом наборе данных (200 записей), интерфейс событий выглядит что-то вроде этого

Но когда я бегу то же самое на большом наборе данных (т.е. я меняю только CASSANDRA_COLUMN_FAMILY), работа никогда не заканчивается внутри терминала, журнал выглядит следующим образом

и после ~ 2 минут, STDERR для исполнителей выглядит так

и после ~ 7 минут, я получаю

Exception in thread "main" java.lang.OutOfMemoryError: GC overhead limit exceeded

в моем терминале, и у меня есть для ручного уничтожения процесса SparkSubmit. Однако большой набор данных был проиндексирован из двоичного файла, который занимал всего 22 МБ, и делая nodetool status, я вижу, что только данные объемом ~ 115 МБ хранятся в обоих моих узлах кассандры. Я также пытался использовать Spark SQL в моем наборе данных, но у меня тоже есть аналогичные результаты. Где я ошибаюсь в своей настройке и что мне делать, чтобы успешно обработать мой набор данных, как для программы Transformation-Action, так и для программы, использующей Spark SQL.

Я уже пробовал следующие методы

• Использование -Xms1G -Xmx1G для увеличения объема памяти, но программа не с исключением того, что я должен вместо этого установить spark.executor.memory, что у меня есть.

• Использование spark.cassandra.input.split.size, что не позволяет утверждать, что это недопустимый вариант, и аналогичная опция spark.cassandra.input.split.size_in_mb, которую я установил в 1, без эффекта.

EDIT

на основе this ответа, я также попытался следующие методы:

• набор spark.storage.memoryFraction до 0

• не установлен spark.storage.memoryFraction к нулю и использования persist с MEMORY_ONLY , MEMORY_ONLY_SER, MEMORY_AND_DISK и MEMORY_AND_DISK_SER.

Версия:

• Спарк: 1.4.0

• Кассандра: 2.1.6

• искровой Cassandra-разъем: 1.4.0-М1

Answer 1

Я думаю, что есть выпуском в новейшем разъеме искро-кассандра. Предполагается, что параметр spark.cassandra.input.split.size_in_mb имеет значение по умолчанию 64 МБ, которое интерпретируется как 64 байта в коде. Это приводит к созданию слишком большого количества разделов, которые не могут быть запланированы с помощью искры. Попробуйте установить значение conf на

spark.cassandra.input.split.size_in_mb=67108864