мне в телеге очень непросто править код(( поэтому смотри, что там со скобками и прочим, но, походу, задача расширилась и теперь это обработка множества задач с последующей суммой результатов

Похоже, надо делать не так, ведь результаты будут некорректные, перенесенная агрегация не обойдется без последствий

Так и есть, да!

Ну да, агрегацию надо после юниона, конечно

А значит и часть преобразований

Вобщем я в мапе только фильтр добавил и схему кейс класса, чтобы сократить данные, а груп бай с аггрегацией после юниона и еще раз схему прилепил:

val aggDeliveryQuality = aggTableParquetList
     .grouped(128)
     .toList
     .map(files => sparkSession.read.parquet(files: _*))
     .map(_.filter(filterClause).as[ViewabilitySourceAgg])
     .reduce(_ union _)
     .groupBy("request_id")
     .agg(sum(sumColumn).as("viewable_impression"))
     .select(colNamesAgg.map(name => col(name)): _*)
     .dropDuplicates()
     .as[ViewabilitySourceAgg]
     .as("ViewabilitySourceAgg")

.toList.par

Начал работать. Сейчас посмотрю как продвигается.. Par забыл. Но пока без него вроде первичный результат может показать что то. План выстроил вроде как требовалось: 128 файлов в каждом датафрейме, юнион и потом у меня джойн еще идет с другим таким же юнионом

Job aborted due to stage failure: Total size of serialized results of 26635 tasks (1024.0 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)

на клиента тащите чтоли

-- conf spark.driver.maxResultSize=0, и на драйвере должно быть достаточно памяти

Да. Сейчас вижу что забыл поменять на кластер в сабмите

даже на кластере драйвер бежит на одной машине

параллельные стримы то зачем

а почему так много датафреймов, хаускипинг процессы сломались?

Мне уже правильно указали что оно и в скале есть вполне себе из коробки

А что в скале ‘параллельные стримы’?

Там параллельные сиевенсы же

я уж думал ты про жавовые. но скаловые пар сиквенсы тож моветон

чтобы начать proof of concept, сойдут и параллельные коллекции ))