Spark：数据集上的映射组

Question

我正在下面的数据集上尝试这个 mapgroups 函数，但不知道为什么我的“总价值”列为 0。我在这里遗漏了什么？？？请指教

Spark 版本 - 2.0 Scala 版本 - 2.11

case class Record(Hour: Int, Category: String,TotalComm: Double, TotalValue: Int)
val ss = (SparkSession)
import ss.implicits._

val df: DataFrame = ss.sparkContext.parallelize(Seq(
(0, "cat26", 30.9, 200), (0, "cat26", 22.1, 100), (0, "cat95", 19.6, 300), (1, "cat4", 1.3, 100),
(1, "cat23", 28.5, 100), (1, "cat4", 26.8, 400), (1, "cat13", 12.6, 250), (1, "cat23", 5.3, 300),
(0, "cat26", 39.6, 30), (2, "cat40", 29.7, 500), (1, "cat4", 27.9, 600), (2, "cat68", 9.8, 100),
(1, "cat23", 35.6, 500))).toDF("Hour", "Category","TotalComm", "TotalValue")

val resultSum = df.as[Record].map(row => ((row.Hour,row.Category),(row.TotalComm,row.TotalValue)))
.groupByKey(_._1).mapGroups{case(k,iter) => (k._1,k._2,iter.map(x => x._2._1).sum,iter.map(y => y._2._2).sum)}
.toDF("KeyHour","KeyCategory","TotalComm","TotalValue").orderBy(asc("KeyHour"))

resultSum.show()

+-------+-----------+---------+----------+
|KeyHour|KeyCategory|TotalComm|TotalValue|
+-------+-----------+---------+----------+
|      0|      cat26|     92.6|         0|
|      0|      cat95|     19.6|         0|
|      1|      cat13|     12.6|         0|
|      1|      cat23|     69.4|         0|
|      1|       cat4|     56.0|         0|
|      2|      cat40|     29.7|         0|
|      2|      cat68|      9.8|         0|
+-------+-----------+---------+----------+  

Answer 1

iter里面mapGroups是一个缓冲区，计算只能执行一次。所以当你 sum asiter.map(x => x._2._1).sum那么iter缓冲区中没有任何东西时，因此 iter.map(y => y._2._2).sum操作产生0。所以你必须找到一种机制来计算在同一次迭代中两者的总和

带有 ListBuffers 的 for 循环

为简单起见，我使用了for循环并ListBuffer同时求和

val resultSum = df.as[Record].map(row => ((row.Hour,row.Category),(row.TotalComm,row.TotalValue)))
  .groupByKey(_._1).mapGroups{case(k,iter) => {
  val listBuffer1 = new ListBuffer[Double]
  val listBuffer2 = new ListBuffer[Int]
      for(a <- iter){
        listBuffer1 += a._2._1
        listBuffer2 += a._2._2
      }
      (k._1, k._2, listBuffer1.sum, listBuffer2.sum)
    }}
  .toDF("KeyHour","KeyCategory","TotalComm","TotalValue").orderBy($"KeyHour".asc)

这应该给你正确的结果

+-------+-----------+---------+----------+
|KeyHour|KeyCategory|TotalComm|TotalValue|
+-------+-----------+---------+----------+
|      0|      cat26|     92.6|       330|
|      0|      cat95|     19.6|       300|
|      1|      cat23|     69.4|       900|
|      1|      cat13|     12.6|       250|
|      1|       cat4|     56.0|      1100|
|      2|      cat68|      9.8|       100|
|      2|      cat40|     29.7|       500|
+-------+-----------+---------+----------+

我希望答案有帮助