А, уже сказали

все же, лучше не я буду приводить, мне сейчас не нужно ^^

Я привел к датасету уже который был до этого с прежним подходом

ваще пушка))

Я приаел схему в конце после фильтров и аггрегаций. Лучше перед сразу после юниона?

А не. Что то так не берет

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

.map(files => sparkSession.read.parquet(files: _*))
.map(_.select(colNamesAgg.map(name => col(name)): _*))
.reduce(_ union _)

навскидку вот так

время то улучшается? или все потрачено?

У меня проблемы с интернетом. Никак джар не зааплодю

(( надеюсь, что будет лучше, но если нет, то advanced техники у меня нет

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")

нет, смысл был в том, чтобы перенести строку кода .select(colNamesAgg.map(name => col(name)): _*) в каждый датафрейм до union

Тогда я groupby не смогу сделать потому что селект берет только пару колонок

Так параллелизм потеряли же, нет?

Лучше тогда как то схему прицепить к каждому юниону, нет?

лучше сразу при чтении укажи

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

Но груп бай так в каждом отдельном может привести к другому результату после юниона. Или я ошибаюсь?