Так сделай join через case

это как?

понял, надо попробовать

DE есть вопрос. При работе приложения на spark, требуется конфигурация толстых экзекуторов 64Gbx16VCPU, но есть джоба у которой всего 10 тасков, на каждую таску по 1 партиции, но партиции жирные порезать не могу. И вот spark садит эти таски на 2 экзеутора, выбирая те которые меньше всего выполняли тасков. И все уходит в тормоза, так как один экзекутор тащит все шафлы к себе и записывает на диск. Есть патерн как сделать так что бы, таски раскидывались на разные экзекуторы ?

Если выходной файл один, то и экзекьютор будет один

Врайтер свой, на каждую партицию.

Поэтому и екзекутор пишет сразу 5 фйлов и тормозит ....

Можно перед записью сделать репартишен

По идее разницы никакой нет, что эти потоки на одном экзекьюторе, что на разных, один поток = 1 ядро

Репартишн сделан, 100 партиций, но он их все посадил на 2 экзекутора. Я уже нашел причину, у меня стоит мин экзекуторов 2 шт, он просто перед распределегием остальных убил.

У меня толстые экщекуторы по 16 ядер.

Накидываю варианты:
1. Поставить minExecutors побольше
2. spark.task.cpus / spark.task.resource.{resourceName}.amount может выкружить больше ресурсов на жирные таски, но не знаю, с какой версии, и поможет ли в этом кейсе. Раскидывание тасков по разным экзекуторам, скорее всего, произойдёт, но не уверен, будет ли лучше! Может, разумнее экзекуторы на эту джобу дать небольшие по одному core?
3. Делать другой репартишен, и делать его сразу перед экшеном, чтобы ресурс менеджер не убивал экзекуторов
Вообще, мне непонятно, сколько же там партиций/тасков? Десять или сто? Если там десять значений, то сто партиций просто так не получится, нужно солить. А если их нельзя посолить и порезать, то этот варик не подойдёт

Спасибо. Там 100 партиций, но в них 10 жирных, разделить нельзя. А как экзекуторы именно к джобе изменить ?

Они сделаны из 30000, убрана соль и собранно в 100.

Эмм, убить сессию и поднять заново на том же контексте, но с другими executor.cores и .memory. Контест тоже можно переподнять. Не вариант, если необходимо переиспользование датасетов

Не не тема, дата сет нужен.

таки это data skew, нужно его внутри резать. Предлагаю склеивать исходные партиции по уму:
1. выбрать жирные  с помощью предварительной джобы, определить, на сколько можно побить каждую, сделать HashMap
2. простой udf добавить поле с солью: functions.ceil(functions.rand()*hashMap.getOrElse(key, 0))

Я их как раз собрал, так как они мне нужны для записи в файл. А так я их порезал до массовой обработки. Чуть позже буду изучать поблочную запичь в hdfs.

Сейчас пока стрим.

И у меня не дата сет на этом этапе, а rdd