SparkSQL: Avg на основе столбца после GroupBy

Question

У меня есть класс студентов, и мне нужно сначала сгруппировать их по первому столбцу, который является университетом, а затем отобразить среднее число учащихся в каждом курсе, как это. Каков самый простой способ сделать этот запрос?

+----------+-------------------+                
|university| avg of students | 
+----------+--------------------+ 
|  MIT| 3    | 
| Cambridge| 2.66 

Вот набор данных.

case class grade(university: String, courseId: Int, studentId: Int, grade: Double) 

val grades = List(grade(
grade("Cambridge", 1, 1001, 4), 
grade("Cambridge", 1, 1004, 4), 
grade("Cambridge", 2, 1006, 3.5), 
grade("Cambridge", 2, 1004, 3.5), 
grade("Cambridge", 2, 1002, 3.5), 
grade("Cambridge", 3, 1006, 3.5), 
grade("Cambridge", 3, 1007, 5), 
grade("Cambridge", 3, 1008, 4.5), 
grade("MIT", 1, 1001, 4), 
grade("MIT", 1, 1002, 4), 
grade("MIT", 1, 1003, 4), 
grade("MIT", 1, 1004, 4), 
grade("MIT", 1, 1005, 3.5), 
grade("MIT", 2, 1009, 2)) 

Answer 1

gradesRdd.map({ case Grade(university: String, courseId: Int, studentId: Int, gpa: Int) => 
    ((university),(courseId))}).mapValues(x => (x, 1)) 
    .reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2)) 
    .mapValues(y => 1.0 * y._1/y._2).collect 
    res73: Array[(String, Double)] = Array((Cambridge,2.125), (MIT,1.1666666666666667)) 

Answer 2

1) Первая группиЙ университет

2) затем получить подсчитывать курс в университет

3) затем группиЙ courseId

4), то получите рассчитывать студент за курс

grades.groupBy(_.university).map { case (k, v) => 
    val courseCount = v.map(_.courseId).distinct.length 
    val studentCountPerCourse = v.groupBy(_.courseId).map { case (k, v) => v.length }.sum 
    k -> (studentCountPerCourse.toDouble/courseCount.toDouble) 
    } 

Scala REPL

scala> val grades = List(
     grade("Cambridge", 1, 1001, 4), 
     grade("Cambridge", 1, 1004, 4), 
     grade("Cambridge", 2, 1006, 3.5), 
     grade("Cambridge", 2, 1004, 3.5), 
     grade("Cambridge", 2, 1002, 3.5), 
     grade("Cambridge", 3, 1006, 3.5), 
     grade("Cambridge", 3, 1007, 5), 
     grade("Cambridge", 3, 1008, 4.5), 
     grade("MIT", 1, 1001, 4), 
     grade("MIT", 1, 1002, 4), 
     grade("MIT", 1, 1003, 4), 
     grade("MIT", 1, 1004, 4), 
     grade("MIT", 1, 1005, 3.5), 
     grade("MIT", 2, 1009, 2)) 
// grades: List[grade] = List(...) 

scala> grades.groupBy(_.university).map { case (k, v) => 
     val courseCount = v.map(_.courseId).distinct.length 
     val studentCountPerCourse = v.groupBy(_.courseId).map { case (k, v) => v.length }.sum 
     k -> (studentCountPerCourse.toDouble/courseCount.toDouble) 
    } 
// res2: Map[String, Double] = Map("MIT" -> 3.0, "Cambridge" -> 2.6666666666666665)