Проблема при преобразовании MS-SQL Query в Spark SQL

Question

Я хочу преобразовать этот базовый SQL-запрос в Spark

select Grade, count(*) * 100.0 / sum(count(*)) over()
from StudentGrades
group by Grade

Я пытался использовать оконные функции в свече, как это

val windowSpec = Window.rangeBetween(Window.unboundedPreceding,Window.unboundedFollowing)

df1.select(
$"Arrest"
).groupBy($"Arrest").agg(sum(count("*")) over windowSpec,count("*")).show()


+------+-------------------------------------------------------------------- 
----------+--------+
|Arrest|sum(count(1)) OVER (RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED 
FOLLOWING)|count(1)|
+------+-------------------------------------------------------------------- 
----------+--------+
|  true|                                                                        
665517|  184964|
| false|                                                                        
665517|  480553|
+------+------------------------------------------------------------------------------+--------+

Но когда я пытаюсь разделить на count (*), это из-за ошибки

df1.select(
$"Arrest"
).groupBy($"Arrest").agg(count("*")/sum(count("*")) over 
windowSpec,count("*")).show()

Не допускается использование агрегатной функции в аргументе другой агрегатной функции. Пожалуйста, используйте внутреннюю статистическую функцию в подзапросе. ;;

Мой вопрос: когда я уже использую count () внутри sum () в первом запросе, я не получаю никаких ошибок при использовании агрегатной функции внутри другой агрегатной функции, но почему возникает ошибка во втором?

Answer 1

Пример:

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

val df = sc.parallelize(Seq(
   ("A", "X", 2, 100), ("A", "X", 7, 100), ("B", "X", 10, 100),
   ("C", "X", 1, 100), ("D", "X", 50, 100), ("E", "X", 30, 100)
    )).toDF("c1", "c2", "Val1", "Val2")

val df2 = df
  .groupBy("c1")
  .agg(sum("Val1").alias("sum"))
  .withColumn("fraction", col("sum") /  sum("sum").over())

df2.show

Вам нужно будет приспособиться к вашей собственной ситуации. Например. считать вместо суммы. Следующим образом:

val df2 = df
  .groupBy("c1")
  .agg(count("*"))
  .withColumn("fraction", col("count(1)") /  sum("count(1)").over())

возвращения:

+---+--------+-------------------+
| c1|count(1)|           fraction|
+---+--------+-------------------+
|  E|       1|0.16666666666666666|
|  B|       1|0.16666666666666666|
|  D|       1|0.16666666666666666|
|  C|       1|0.16666666666666666|
|  A|       2| 0.3333333333333333|
+---+--------+-------------------+

Вы можете сделать х 100. Я заметил, что псевдоним, кажется, не работает в соответствии с суммой, так что обойти это и оставил сравнение выше. Опять же, вам нужно будет адаптировать свою специфику, это часть моих общих модулей для исследований и тому подобное.