问题描述
我有三个字符串类型的数组，其中包含以下信息:
I have three Arrays of string type containing following information:
- groupBy数组:包含我要对数据进行分组的列的名称.
- 聚合数组:包含我要聚合的列的名称.
- operations数组:包含我要执行的聚合操作
我正在尝试使用spark数据帧来实现此目的. Spark数据帧提供了agg()，您可以在其中传递Map [String，String](具有列名和相应的聚合操作)作为输入，但是我想对数据的同一列执行不同的聚合操作.关于如何实现这一目标的任何建议?
I am trying to use spark data frames to achieve this. Spark data frames provide an agg() where you can pass a Map [String,String] (of column name and respective aggregate operation ) as input, however I want to perform different aggregation operations on the same column of the data. Any suggestions on how to achieve this?
推荐答案
斯卡拉:
例如，您可以从名称到功能映射具有定义的mapping的功能列表:
You can for example map over a list of functions with a defined mapping from name to function:
import org.apache.spark.sql.functions.{col, min, max, mean}
import org.apache.spark.sql.Column
val df = Seq((1L, 3.0), (1L, 3.0), (2L, -5.0)).toDF("k", "v")
val mapping: Map[String, Column => Column] = Map(
"min" -> min, "max" -> max, "mean" -> avg)
val groupBy = Seq("k")
val aggregate = Seq("v")
val operations = Seq("min", "max", "mean")
val exprs = aggregate.flatMap(c => operations .map(f => mapping(f)(col(c))))
df.groupBy(groupBy.map(col): _*).agg(exprs.head, exprs.tail: _*).show
// +---+------+------+------+
// | k|min(v)|max(v)|avg(v)|
// +---+------+------+------+
// | 1| 3.0| 3.0| 3.0|
// | 2| -5.0| -5.0| -5.0|
// +---+------+------+------+
或
df.groupBy(groupBy.head, groupBy.tail: _*).agg(exprs.head, exprs.tail: _*).show
不幸的是，内部公开使用的解析器SQLContext并未公开公开，但您始终可以尝试构建普通的SQL查询:
Unfortunately parser which is used internally SQLContext is not exposed publicly but you can always try to build plain SQL queries:
df.registerTempTable("df")
val groupExprs = groupBy.mkString(",")
val aggExprs = aggregate.flatMap(c => operations.map(
f => s"$f($c) AS ${c}_${f}")
).mkString(",")
sqlContext.sql(s"SELECT $groupExprs, $aggExprs FROM df GROUP BY $groupExprs")
Python :
from pyspark.sql.functions import mean, sum, max, col
df = sc.parallelize([(1, 3.0), (1, 3.0), (2, -5.0)]).toDF(["k", "v"])
groupBy = ["k"]
aggregate = ["v"]
funs = [mean, sum, max]
exprs = [f(col(c)) for f in funs for c in aggregate]
# or equivalent df.groupby(groupBy).agg(*exprs)
df.groupby(*groupBy).agg(*exprs)
另请参阅:
这篇关于在spark数据帧的同一列上进行多个聚合操作的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！