Spark 2.0 DataSets groupByKey 和 除法操作以及类型安全

我对 Spark 2.0 DataSets 非常满意，因为它的编译时类型安全。但这里有几个我无法解决的问题，我也没有找到很好的文档。

问题#1 - 对聚合列进行除法运算 -考虑下面的代码 - 我有一个 DataSet[MyCaseClass]，我想对 c1、c2、c3 和 sum(c4) / 8 进行 groupByKey。如果我只计算总和，下面的代码效果很好，但它给出了 diverge(8) 的编译时错误。我想知道我怎样才能实现以下目标。

final case class MyClass (c1: String,
                          c2: String,
                          c3: String,
                          c4: Double)

    val myCaseClass: DataSet[MyCaseClass] = ??? // assume it's being loaded

    import sparkSession.implicits._
    import org.apache.spark.sql.expressions.scalalang.typed.{sum => typedSum}

     myCaseClass.
       groupByKey(myCaseClass =>
          (myCaseClass.c1, myCaseClass.c2, myCaseClass.c3)).
          agg(typedSum[MyCaseClass](_.c4).name("sum(c4)").
          divide(8)). //this is breaking with exception
       show()

如果我删除 .divide(8) 操作并运行上面的命令，它会给出以下输出。

+-----------+-------------+
|        key|sum(c4)      |
+-----------+-------------+
| [A1,F2,S1]|         80.0|
| [A1,F1,S1]|         40.0|  
+-----------+-------------+

问题 #2 - 将 groupedByKey 结果转换为另一个类型化 DataFrame -现在我的问题的第二部分是我想再次输出一个类型化的数据集。为此，我有另一个案例类（不确定是否需要），但我不确定如何映射分组结果 -

final case class AnotherClass(c1: String,
                          c2: String,
                          c3: String,
                          average: Double) 

 myCaseClass.
           groupByKey(myCaseClass =>
              (myCaseClass.c1, myCaseClass.c2, myCaseClass.c3)).
              agg(typedSum[MyCaseClass](_.c4).name("sum(c4)")).
as[AnotherClass] //this is breaking with exception

但这再次失败，并出现异常，因为按键结果分组未直接与 AnotherClass 映射。

PS：非常欢迎任何其他实现上述目标的解决方案。

第一个问题可以通过一直使用类型化列来解决（KeyValueGroupedDataset.agg期望TypedColumn(-s)） 您可以将聚合结果定义为：

val eight = lit(8.0)
  .as[Double]  // Not necessary

val sumByEight = typedSum[MyClass](_.c4)
  .divide(eight)
  .as[Double]  // Required
  .name("div(sum(c4), 8)")

并将其插入以下代码：

val myCaseClass = Seq(
  MyClass("a", "b", "c", 2.0),
  MyClass("a", "b", "c", 3.0)
).toDS

myCaseClass
  .groupByKey(myCaseClass => (myCaseClass.c1, myCaseClass.c2, myCaseClass.c3))
  .agg(sumByEight)

to get

+-------+---------------+
|    key|div(sum(c4), 8)|
+-------+---------------+
|[a,b,c]|          0.625|
+-------+---------------+

第二个问题是由于使用了不符合数据形状的类而导致的。正确的表示可能是：

case class AnotherClass(key: (String, String, String), sum: Double)

与上面定义的数据一起使用：

 myCaseClass
   .groupByKey(myCaseClass => (myCaseClass.c1, myCaseClass.c2, myCaseClass.c3))
   .agg(typedSum[MyClass](_.c4).name("sum"))
   .as[AnotherClass]

会给出：

+-------+---+
|    key|sum|
+-------+---+
|[a,b,c]|5.0|
+-------+---+

but .as[AnotherClass]如果这里没有必要Dataset[((String, String, String), Double)]是可以接受的。

你当然可以跳过所有这些，然后mapGroups（尽管并非没有性能损失）：

import shapeless.syntax.std.tuple._   // A little bit of shapeless

val tuples = myCaseClass
 .groupByKey(myCaseClass => (myCaseClass.c1, myCaseClass.c2, myCaseClass.c3))
 .mapGroups((group, iter) => group :+ iter.map(_.c4).sum)

有结果

+---+---+---+---+   
| _1| _2| _3| _4|
+---+---+---+---+
|  a|  b|  c|5.0|
+---+---+---+---+

reduceGroups可能是一个更好的选择：

myCaseClass
  .groupByKey(myCaseClass => (myCaseClass.c1, myCaseClass.c2, myCaseClass.c3))
  .reduceGroups((x, y) => x.copy(c4=x.c4 + y.c4))

由此产生的Dataset:

+-------+-----------+    
|     _1|         _2|
+-------+-----------+
|[a,b,c]|[a,b,c,5.0]|
+-------+-----------+