Тупой вопрос - как эмулировать df.partitionBy(col) если мне нужно записать каждую запись в редис как json?

Использую spark-redis, который записывает датафрейм в редис как отдельные чанки. Код, который читает записанные данные использует обычный питоновский драйвер, то есть придется писать логику сборки, с чем не хочется заморачиваться.

Отличный канал! Понятнее и быстрее, чем доку изучать

xzljv lasvjt has been banned! Reason: CAS ban.

товарищи, научите работать спарком/spark-shell. понадобилось посравнивать две, что-то типа EAV таблички, одна из бэкапа, другая свежая. таблички по 50-70 млрд строк. даже exceptAll(), генерирующий HashJoin падает, пришлось задирать spark.executor.memory до 6G и memoryOverhead до 3G. в запросах где делаю left join и план с MergeSortJoin и этого не хватает.
может есть какой параметр ограничивающий сортировки cпарку ? как вообще думана работа спарк, допустим из зепелина или spark on hive. задачи то разные случаются, я же не буду подстраивать executor.memory/memoryOverhead под каждую задачу.

Это нормально, когда тяжёлые расчёты имеют свои конфиги

спарк не должен падать по памяти, максимум долгий спилл на диск, вероятно, причина в перекосе данных на 1 ключ

в том и дело что запросы что я пытаюсь выполнить не должны перекоса давать

exceptAll() - я так понимаю проблема могла бы быть если бы миллионы полных дублей обнаружились. но такого в данных нет

как вариант, может left anti join сделать по ключу?

пробовал по разному, и anti и MINUS в sql синтаксисе. все что не exceptAll дает MergeSortJoin и тупит часами, а в части executors вижу
#
# java.lang.OutOfMemoryError: Java heap space
# -XX:OnOutOfMemoryError="kill %p"
#   Executing /bin/sh -c "kill 129217"...
2021-05-12 04:29:13 ERROR [SIGTERM handler] CoarseGrainedExecutorBackend:43 - RECEIVED SIGNAL TERM

мерж join дефолтный алгоритм в спарк, так что это норма для lj. Посмотрите в мониторинге самого спарка, сколько строк в каждый таск попадает, наверняка у кого-то много, а у кого-то мало. Тут или памяти добавлять или бороться за равномерное распределение

ну тогда выходит что аналитики толком ни в зепелине ни в hive on spark не могут работать, без тонкой настройки под каждого

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

А сколько оперативы на одно ядро экзекутора?

И сколько миллионов строк в среднем на таск?

4-6G на executor + 3G memoryOverhead, ну и порядка 40 executors ставлю

Чаще всего нужно просто увеличить spark.sql.shuffle.partitions, и всё отработает, если только нет какой-нибудь одной таски с кучей записей с одним и тем же ключом

А сколько тасков отрабатывает?

199/200?