Эффективные операции с искровым набором данных при разделении по перекрывающимся столбцам

Чтобы заполнить agtM последним непустым значением agt, вы можете использовать last("agt", ignoreNulls) с rowsBetween() для w2:

val ds = Seq(
  ("a", "1", ""),
  ("b", "2", "4"),
  ("a", "1", "3"),
  ("b", "2", "4"),
  ("c", "1", "3"),
  ("a", "6", "8"),
  ("b", "2", "4"),
  ("a", "1", "4")
).toDF("guid", "timestamp", "agt").
  as[(String, String, String)]

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window

val w = Window.partitionBy($"guid")
val w1 = Window.partitionBy($"guid", $"timestamp")
val w2 = Window.partitionBy($"guid").orderBy($"timestamp".desc).
  rowsBetween(Window.unboundedPreceding, 0)

ds.
  withColumn("minimum", min("timestamp").over(w)).
  withColumn("count", count("*").over(w1)).
  withColumn("agt", when($"agt" =!= "", $"agt")).
  withColumn("agtM", last("agt", ignoreNulls = true).over(w2)).
  na.fill("", Seq("agt")).
  dropDuplicates("guid", "timestamp").
  show
// +----+---------+---+-------+-----+----+
// |guid|timestamp|agt|minimum|count|agtM|
// +----+---------+---+-------+-----+----+
// |   c|        1|  3|      1|    1|   3|
// |   b|        2|  4|      2|    3|   4|
// |   a|        1|   |      1|    3|   8|
// |   a|        6|  8|      1|    1|   8|
// +----+---------+---+-------+-----+----+

Учитывая, что каждая из ваших спецификаций окна w, w1 и w2 имеет свои особые требования, я не уверен, что можно многое сделать для уменьшения перетасовки. Вы можете изучить неоконные подходы, хотя результирующий набор данных, который вы намереваетесь создать, хорошо подходит для использования оконных функций.

Мне нужно вычислить
минимальную метку времени для каждой строки при группировке по guid.
Количество для каждого ключа при группировке по (guid, timestamp)
agtM строки при группировке по guid и упорядочении по отметке времени (описание), а затем взять сначала непустое значение, иначе ""

Глядя на ваше требование, вам нужно вычислить минимум временной метки, agtM (последний) agt для группы guid и подсчитать при группировке по guid и временной метке. Эти требования предполагают, что вам потребуются три группировки и три перетасовки.

Первая группировка и перетасовка - чтобы найти количество

val dfWithCount = df
      .groupBy("guid", "timestamp")
      .agg(count("guid").as("count"))

Вторая и третья группировка и перемешивание

последний agt, т. е. agtM можно найти с помощью функции Window, а минимальную метку времени можно найти с помощью других groupBy и aggregation

val dfWithMinAndMax = df.withColumn("agtM", first("agt").over(windowSpec))
      .groupBy("guid", "agtM")
      .agg(min("timestamp").as("minimum")
      )

Наконец, вы join оба фрейма данных

val finalDF = dfWithCount.join(dfWithMinAndMax, Seq("guid"))

Это даст вам правильный фрейм данных но без agt

+----+---------+-----+----+-------+
|guid|timestamp|count|agtM|minimum|
+----+---------+-----+----+-------+
|c   |1        |1    |3   |1      |
|b   |2        |3    |4   |2      |
|a   |1        |3    |8   |1      |
|a   |6        |1    |8   |1      |
+----+---------+-----+----+-------+

Я предполагаю, что agt не так важно, но если вам это действительно нужно, вам понадобится еще одна группировка и перетасовка и объединение

val dfWithAgt = df.groupBy("guid", "timestamp").agg(min("agt").as("agt"))

finalDF.join(dfWithAgt, Seq("guid", "timestamp"))

что даст тебе

+----+---------+-----+----+-------+---+
|guid|timestamp|count|agtM|minimum|agt|
+----+---------+-----+----+-------+---+
|c   |1        |1    |3   |1      |3  |
|b   |2        |3    |4   |2      |4  |
|a   |1        |3    |8   |1      |   |
|a   |6        |1    |8   |1      |8  |
+----+---------+-----+----+-------+---+

Порядок столбцов можно задать с помощью select.

надеюсь ответ будет полезен

Если сначала разбить его по guid, а затем работать с итераторами, логически получится меньше перетасовки. Не уверен в эффекте, если данные внутри каждой группы огромны.

df.toDS().groupByKey(_.guid).flatMapGroups((a,b) => {
          val list = b.toList
          val minimum = list.minBy(_.timestamp).timestamp
          val filteredList = list.filterNot(_.agt == "")
          val agtM = if(filteredList.isEmpty) "" else filteredList.maxBy(_.timestamp).agt
          list.groupBy(_.timestamp).map(r => (r._2.head.guid, r._1, r._2.head.agt, minimum, r._2.length, agtM))
        }).select($"_1".as("guid"), $"_2".as("timestamp"),
          $"_3".as("agt"), $"_4".as("minimum"), $"_5".as("count"), $"_6".as("agtM")).show()