如何命名聚合列？

Question

我在Scala中使用Spark,我的聚合列是匿名的.有没有一种方便的方法来重命名数据集中的多个列？我想到了征收模式与as更关键的列是一个结构(由于groupBy操作),我不能找出如何定义case class与StructType它.

我尝试按如下方式定义模式:

val returnSchema = StructType(StructField("edge", StructType(StructField("src", IntegerType, true),
                                                             StructField("dst", IntegerType), true)), 
                              StructField("count", LongType, true))
edge_count.as[returnSchema]

但是我遇到了编译错误:

Message: <console>:74: error: overloaded method value apply with alternatives:
  (fields: Array[org.apache.spark.sql.types.StructField])org.apache.spark.sql.types.StructType <and>
  (fields: java.util.List[org.apache.spark.sql.types.StructField])org.apache.spark.sql.types.StructType <and>
  (fields: Seq[org.apache.spark.sql.types.StructField])org.apache.spark.sql.types.StructType
 cannot be applied to (org.apache.spark.sql.types.StructField, org.apache.spark.sql.types.StructField, Boolean)
       val returnSchema = StructType(StructField("edge", StructType(StructField("src", IntegerType, true),

Answer 1

最好的解决方案是明确命名列,例如,

df
  .groupBy('a, 'b)
  .agg(
    expr("count(*) as cnt"),
    expr("sum(x) as x"),
    expr("sum(y)").as("y")
  )

如果您使用的是数据集,则必须提供列的类型,例如expr("count(*) as cnt").as[Long].

您可以直接使用DSL,但我经常发现它比简单的SQL表达式更冗长.

如果要进行批量重命名,请使用a Map然后foldLeft使用数据框.

Answer 2

我最终使用aliases 来select声明；例如，

ds.select($"key.src".as[Short], 
          $"key.dst".as[Short], 
          $"sum(count)".alias("count").as[Long])

首先，我必须用来printSchema确定派生列名称：

> ds.printSchema

root
 |-- key: struct (nullable = false)
 |    |-- src: short (nullable = false)
 |    |-- dst: short (nullable = false)
 |-- sum(count): long (nullable = true)