作者:霞霞123321 | 来源:互联网 | 2023-01-29 20:19
使用Scala Spark,如何使用类型化的数据集API舍入聚合列?
另外,如何通过groupby操作保留数据集的类型?
这是我目前拥有的:
case class MyRow(
k1: String,
k2: String,
c1: Double,
c2: Double
)
def groupTyped(ds: Dataset[MyRow]): Dataset[MyRow] = {
import org.apache.spark.sql.expressions.scalalang.typed._
ds.groupByKey(row => (row.k1, row.k2))
.agg(
avg(_.c1),
avg(_.c2)
)
.map(r => MyRow(r._1._1, r._1._2, r._2, r._3))
}
如果我将替换为avg(_.c1)
,则会round(avg(_.c1))
出现类型错误。四舍五入的正确方法是什么?
这.map(...)
行感觉不对-有没有更优雅的方法来保留我的数据集的类型?
谢谢!
1> Ahe..:
虽然可接受的答案有效且更笼统,但在这种情况下,您也可以使用回合。您只需要在四舍五入后使用来键入列.as[T]
(也需要将类型定义为avg)。
.agg(
// Alternative ways to define a type to avg
round(avg((r: MyRow) => r.c1)).as[Double],
round(avg[MyRow](_.c2)).as[Double]
)