将火花数据帧中的多行合并到一行中
问题描述:
我有一个包含2列的数据帧:timestamp,值 timestamp是自时间以来的时间,值是浮点值。 我想合并行的平均值分钟。 这意味着我想要获取时间戳来自同一轮分钟(自纪元以来60秒的时间间隔)的所有行,并将它们合并到单个行中,其中值列将是所有值的平均值。将火花数据帧中的多行合并到一行中
举一个例子,让我们假设我的数据帧看起来像这样:
timestamp value
--------- -----
1441637160 10.0
1441637170 20.0
1441637180 30.0
1441637210 40.0
1441637220 10.0
1441637230 0.0
第一4行是相同的分的一部分(1441637160%60 == 0,1441637160 + 60 == 1441637220) 最后2行是另一分钟的一部分。 我想合并相同分钟的所有行。得到如下结果:
timestamp value
--------- -----
1441637160 25.0 (since (10+20+30+40)/4 = 25)
1441637220 5.0 (since (10+0)/2 = 5)
这样做的最佳方法是什么?
答
您可以简单地进行分组和聚合。随着数据为:
val df = sc.parallelize(Seq(
(1441637160, 10.0),
(1441637170, 20.0),
(1441637180, 30.0),
(1441637210, 40.0),
(1441637220, 10.0),
(1441637230, 0.0))).toDF("timestamp", "value")
进口所需的功能和类:
import org.apache.spark.sql.functions.{lit, floor}
import org.apache.spark.sql.types.IntegerType
创建间隔柱:
val tsGroup = (floor($"timestamp"/lit(60)) * lit(60))
.cast(IntegerType)
.alias("timestamp")
,并用它来进行聚合:
df.groupBy(tsGroup).agg(mean($"value").alias("value")).show
// +----------+-----+
// | timestamp|value|
// +----------+-----+
// |1441637160| 25.0|
// |1441637220| 5.0|
// +----------+-----+
答
首先将时间戳映射到分钟桶,然后使用groupByKey计算平均值。例如:
rdd.map(x=>{val round = x._1%60; (x._1-round, x._2);})
.groupByKey
.map(x=>(x._1, (x._2.sum.toDouble/x._2.size)))
.collect()