Первая и последняя агрегированные функции Spark SQL - неожиданный результат

Question

Получение неожиданного результата при выполнении первой и последней агрегированных функций в Spark Dataframe.

У меня есть кадр данных искры, имеющий столбцы colA, colB, colC, colD, colE, extraCol1, extraCol2

И мне нужно сделать агрегацию на этом фрейме данных к

группировка -> colA и colB, макс. -> colC, макс. -> colD, первая -> colE, extraCol1, extraCol2

Ниже приведен фрейм данных (df), который я использую, и использую искровое разбиение (3)

colA    colB    colC    colD    colE    extraCol1   extracol2
Harshit 23        43    44         A           q    z
Mohit   24        56    62         B           w    x
Harshit 23        32    44         C           e    c
Kali    10        20    460        D           r    v
Aman    20        30    180        E           t    b
Ram     30        100   270        F          yu    n
Kali    10        600   360        G          io    m
Kali    10        600   460        k           p    o

Ниже приведен код scala и spark, который я использую для выполнения операции groupBy

 val cols = List("colA","colB")

 var  aggFuncSeq = List(max(`colC`) as colC_new, max(`colD`) as colD_new, first(`colE`,true) as colE, first(`extracol2`,true) as extracol2, first(`extraCol1`,true) as extraCol1)

 var aggFuncs = aggFuncSeq.map(e => expr(e))

 df = df.groupBy(cols.head, cols.tail: _*).agg(aggFuncs.head, aggFuncs.tail: _*)

 df.show(10)

После выполнения я получаю неожиданный результат, как показано ниже.

colA    colB    colC_new    colD_new    colE    extracol2   extraCol1
Harshit     23      43            44        C       c       e
Aman        20      30            180       E       b       t
Kali        10      600           460       D       v       r
Ram         30      100           270       F       n       yu
Mohit       24      56            62        B       x       w

Но согласно выполненному условию группировки и операции агрегации, выходной результат должен иметь первую строку, соответствующую Harshit, для colE, extracol2, extracol1

Итак, ожидаемый результат был как ниже

colA    colB    colC_new    colD_new    colE    extracol2   extraCol1
Harshit     23      43            44        A       q       z
Aman        20      30            180       E       b       t
Kali        10      600           460       D       v       r
Ram         30      100           270       F       n       yu
Mohit       24      56            62        B       x       w

Но я не могу понять эту концепцию SQL, то, как это возможно. Так что, если кто-нибудь может помочь мне разобраться в этой странной проблеме.

Это из-за разбиения?

Как это дает этот результат и как исправить его как ожидаемый результат?

Любая помощь приветствуется. Спасибо

Answer 1

 import org.apache.spark.sql.functions.{max, _}
    import spark.implicits._

    val columnsDF = Seq(
      ("Harshit", 23, 43, 44, "A", "q", "z"),
      ("Mohit", 24, 56, 62, "B", "w", "x"),
      ("Harshit", 23, 32, 44, "C", "e", "c"),
      ("Kali", 10, 20, 460, "D", "r", "v"),
      ("Aman", 20, 30, 180, "E", "t", "b"),
      ("Ram", 30, 100, 270, "F", "yu", "n"),
      ("Kali", 10, 600, 360, "G", "io", "m"),
      ("Kali", 10, 600, 460, "k", "p", "o")
    ).toDF("ColA", "ColB", "ColC", "ColD", "ColE", "extraCol1", "extraCol2")


    println("Before Aggregation")
    columnsDF.show()

    val cols = List("colA", "colB")

    println("After Aggregation")
    val aggSeqFunction = columnsDF.agg(max(columnsDF.columns(2)),
      max(columnsDF.columns(3)),
      first(columnsDF.columns(4)),
      first(columnsDF.columns(6)),
      first(columnsDF.columns(5)))

    val aggFunction = aggSeqFunction.columns.map(en => expr(en))


    columnsDF.groupBy(cols.head, cols.tail: _*).agg(aggFunction.head, aggFunction.tail: _*).show()

    /*
            +-------+----+---------+---------+------------------+-----------------------+-----------------------+
            |   colA|colB|max(ColC)|max(ColD)|first(ColE, false)|first(extraCol2, false)|first(extraCol1, false)|
            +-------+----+---------+---------+------------------+-----------------------+-----------------------+
            |Harshit|  23|       43|       44|                 A|                      z|                      q|
            |   Aman|  20|       30|      180|                 E|                      b|                      t|
            |   Kali|  10|      600|      460|                 D|                      v|                      r|
            |    Ram|  30|      100|      270|                 F|                      n|                     yu|
            |  Mohit|  24|       56|       62|                 B|                      x|                      w|
            +-------+----+---------+---------+------------------+-----------------------+-----------------------+
     */

Я могу получить ожидаемый результат.

Надеюсь, это поможет.

Answer 2

Когда вы groupBy в Spark, вы можете изменить порядок вашего DataFrame. Но не всегда (например, если ваши данные содержатся на одном работнике, они не изменятся). Следовательно, просто чтобы убедиться и иметь масштабируемое решение, вам нужно изменить порядок в вашей оконной функции.

В этом случае попробуйте это:

val w = Window.partitionBy($"key").orderBy($"value")
df
  .withColumn("row_number", row_number.over(w))
  .where($"row_number" === 1)
  .drop("row_number")

Выбирает только первую строку, отфильтрованную по row_number с row_number, определенным как индекс строки после упорядочения. Впоследствии это отбрасывается, потому что становится бесполезным.

Примечание: вы можете заменить операторы $ операторами col . Это всего лишь ярлык для более краткого кода.