Время создания заданий Spark очень большое, с множеством условий фильтрации на фрейме данных

Question

У меня есть фрейм данных PySpark с формой (1e10, 14), и я хотел бы отфильтровать его с помощью примерно 50 составных операторов OR, т.е.:

sql_string = "
(col1='val1' and col2=5) or 
(col1='val2' and col2=7) or
(col1='val3' and col2=5) or
...
"
df_f = df.filter(sql_string)
df_f.limit(1000).show()

Если число этих отдельных операторов OR составляет <10 , Spark Jobs для метода show создаются мгновенно. <br>Однако, с примерно 15 операционными процессами, для создания Spark Jobs уже требуется около 30 секунд.
И примерно при 20 OR, время для создания любых Spark Jobs увеличивается неуправляемый (более часа).

Начиная примерно с 15 OR, G C Сообщения о распределении отображаются каждые несколько секунд, т.е. на. Похоже на проблему, когда один переходит на Spark Dataframes?

Драйвер имеет 32 ГБ ОЗУ (используется 10 ГБ) и 4 ядра (1 ядро ​​используется 100%, другие около 0%). Ввод / вывод практически равен нулю.
Хотя на одном ядре используется 100%, кластер считает его неактивным, поскольку он завершает работу после установленного мной времени бездействия.

Вот ссылка на план выполнения: https://pastebin.com/7MEv5Sq2.

Answer 1

Исходя из опыта работы со стеком Cloudera, я склонялся к работе с паркетом и куду.

Тем не менее, даже если вы не уверены, что вы спрашиваете в реальности, это больше похоже на наблюдения:

1. 50 фильтров занимает время, никакой фильтр не находится на другом конце спектра, но, очевидно, занимает в этом контексте время обработки, близкое к нулю. ИЛИ обработка стоит дороже.

2. pu sh - снижение очевидно из физического плана и теперь является искрой по умолчанию для обработки ИЛИ C.

3. ИЛИ C двигатель выполняет работу на * sh, поэтому вы наблюдаете гораздо меньшую активность исполнителя.

4. ограничение не может быть перенесено в базу данных или паркет / или c.

5. G C материал можно игнорировать.

Мой общий вывод: с вертикальным элементом формы 1e10, на мой взгляд, нет ничего необычного.

Первый опубликованный ответ неверен.

Answer 2

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

Итак, когда мы кэшируем фрейм данных, он уже имеет его в памяти, и, следовательно, выполняет его быстрее, передавая кэшированный фрейм данных всем исполнителям.

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

Надеюсь, это поможет.