Этот граф можно посмотреть в spark ui в разделе SQL, там можно выйти на описание стейджей с указанием структур, откуда читаются данные. Точно видно название таблицы hive, из которой прочитан датафрейм, скорее всего будет видно и каталог hdfs

sparkContext.setJobGroup прекрасно помогает описать джобы в UI, но для отдельных стейджей неприменима, потому что это описание присваивается из пользовательского кода, а он не имеет всей информации. Когда джоба выполняется нет возможности сказать, что закончен один стейдж или начат другой, ведь оптимизатор может заметно изменить стейджи, планировщик способен пропустить некоторые или повторить их при сбое, и это поведение может быть очень сложно или невозможно предсказать.
Есть что-то похожее на исключение из этого правила: когда читаются файлы из HDFS в InMemoryFileIndex (неленивая часть, всегда выполняется отдельно, потому что требуется получить схему и список исходных файлов и в дальнейшем использовать, например в кодогенерации), description изменяется, чтобы показать, сколько будет просматриваться корневых путей, затем восстанавливается предыдущий description. Впрочем, для отдельных стейджей также неприменимо.

Есть вариант вывести df.queryExecution.executedPlan.verboseString (или другое строковое представление) в лог приложения и по нему дополнительно смотреть

спасибо за подробный ответ

а кто-то может обьяснить как для бабушки, какая разница между
rePartition/coalesce
И зачем они нужно?
Я понял, что repartition что-то раскладывает физически, а coalesce тоже самое но в памяти

coalesce объединяет партишены

repartition вычисляет хеши, и разбивает данные на партишены

по хешам равномерненько

относительно

а в каких случаях что используют  и зачем? ))

например, с помощью coalesce можно сохранить данные одним файлом, а не как обычно

можно запараллелить какие-то операции нужным образом

если у кого-то сотня ядер свободных - разбить RDD на сто партиций и каждую отработать своим ядрышком

окей, а вот такой случай
DataSink0 = (
   df
   .write
.repartition(6)
   .mode("overwrite")
   .format("parquet")
   .partitionBy('partner_id', 'years', 'months', 'days', 'dates')
   .save('s3://datalake-analyst/fund/bonus_sub_fund_txn_tbl')
   )
Есть ли смысл делать и зачем?

запараллелить запись

если хватает экзекьюторов, то может писать в шесть потоков

в моем случае если оставить repartition, то скорость записи увеличивается с 30 мин до 130 мин

можете попробовать - запустить приложение и посмотреть в UI в сколько потоков пойдёт работа

значит не хватает экзекьюторов

накладные расходы на репартишен всегда есть - подсчёт хэша и выборка по хешу

ещё эффект - файлики станут меньше, экономия на памяти при обработке