Как оптимизировать операции spark sql на большом фрейме данных?

У меня есть большая таблица улья (~ 9 миллиардов записей и ~ 45 ГБ в формате орков). Я использую Spark sql для профилирования таблицы, но на выполнение каких-либо операций с этим уходит слишком много времени. Просто подсчет самого фрейма входных данных занимает ~ 11 минут. На выполнение min, max и avg для любого столбца уходит более полутора часов.

Я работаю над кластером с ограниченными ресурсами (поскольку он единственный доступный), всего 9 исполнителей, каждый с 2 ​​ядрами и 5 ГБ памяти на каждого исполнителя, распределенных по 3 физическим узлам.

Есть ли способ оптимизировать это, скажем, сократить время выполнения всех агрегатных функций в каждом столбце до менее 30 минут, по крайней мере, с тем же кластером, или увеличение моих ресурсов - единственный способ? что я лично не очень хочу делать. Одно из решений, с которыми я столкнулся для ускорения операций с фреймами данных, - это их кеширование. Но я не думаю, что это реальный вариант в моем случае.

Во всех реальных сценариях, с которыми я сталкивался, для такого рода нагрузки используются огромные кластеры.

Любая помощь приветствуется. Я использую Spark 1.6.0 в автономном режиме с сериализатором kryo.


apache-spark apache-spark-sql spark-hive spark-dataframe apache-spark-1.6
person aladeen    schedule 10.07.2017    source источник
comment
Можете ли вы добавить код и какую конфигурацию вы используете для выполнения задания? Как таблица разделена, если вообще?   -  person morfious902002    schedule 10.07.2017
comment
@ morfious902002 Я уже упоминал свое окружение. И я запускаю dataframe.count () или hiveContext.sql (выберите max (столбец), min (столбец), avg (столбец) из таблицы). И таблица улья не разбита на разделы, но разделена с использованием столбца, который содержит имя параметра.   -  person aladeen    schedule 10.07.2017
comment
Попробуйте 3 исполнителя по 3-5 ядер каждый и 15 ГБ памяти. Для кеширования вы можете попробовать .persist (StorageLevel.MEMORY_AND_DISK_SER ())   -  person morfious902002    schedule 10.07.2017
comment
@ morfious902002 Я уже пытался сохранить файл memeor_and_disk_ser, который сам по себе занимает около 90 минут, и большая часть идет на диск, поскольку объем оперативной памяти составляет 45 ГБ. И мало помогает. Хотя это полезно для меньшего количества данных, когда они могут полностью уместиться в памяти. И я попробую объединить исполнителей с более крупными, спасибо. Но не могли бы вы объяснить, чем это полезно.   -  person aladeen    schedule 11.07.2017
comment
У вас слишком много мелких исполнителей по сравнению с физическими машинами. Это, в свою очередь, приводит к увеличению накладных расходов. Кроме того, попробуйте выполнить фильтрацию, выбрав нужные столбцы (если это еще не сделано) перед использованием функций.   -  person morfious902002    schedule 11.07.2017
comment
@ morfious902002 Я объединил своих исполнителей, и это не очень помогает, и в моей таблице всего 4 столбца, все, что необходимо, и при выполнении запроса агрегации я запускаю только один столбец за раз.   -  person aladeen    schedule 11.07.2017

Ответы (1)


arrow_upward
1
arrow_downward

В sparkSQL есть несколько интересных функций, например:

Кластеризация / Распределение / Сортировать

Spark позволяет писать запросы на SQL-подобном языке - HiveQL. HiveQL позволяет вам управлять разделением данных, так же, как мы можем использовать это в запросах SparkSQL.

Распространить по

В Spark Dataframe разделен некоторым выражением, все строки, для которых это выражение равно, находятся в одном разделе.

SET spark.sql.shuffle.partitions = 2
SELECT * FROM df DISTRIBUTE BY KEY

Итак, посмотрите, как это работает:

par1: [(1,c), (3,b)]
par2: [(3,c), (1,b), (3,d)]
par3: [(3,a),(2,a)]

Это превратится в:

par1: [(1,c), (3,b), (3,c), (1,b), (3,d), (3,a)]
par2: [(2,a)]

Сортировать по

SELECT * FROM df SORT BY key

в этом случае это будет выглядеть так:

par1: [(1,c),  (1,b), (3,b), (3,c), (3,d), (3,a)]
par2: [(2,a)]

Кластеризовать по

Это ярлык для использования совместно используемого распределения и сортировки по одному и тому же набору выражений.

SET spark.sql.shuffle.partitions =2
SELECT * FROM df CLUSTER BY key

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

person arglee    schedule 12.07.2017
comment
Сообщите мне, помогло ли это ?? - person arglee; 14.07.2017