我有一个数据帧,如下所示:
+-----+---+-----+
| id |ind| freq|
+-----+---+-----+
|user1| 1| 5|
|user2| 0| 13|
|user2| 2| 4|
|user3| 2| 7|
|user3| 3| 45|
+-----+---+-----+
列 ind
具有介于 0 和 3 之间的整数值。
我想为每个用户添加缺少的ind
值,同时用默认值(例如 0(填充freq
列,因此输出数据帧如下所示:
+-----+---+-----+
| id |ind| freq|
+-----+---+-----+
|user1| 0| 0|
|user1| 1| 5|
|user1| 2| 0|
|user1| 3| 0|
|user2| 0| 13|
|user2| 1| 0|
|user2| 2| 4|
|user2| 3| 0|
|user3| 0| 0|
|user3| 1| 0|
|user3| 2| 7|
|user3| 3| 45|
+-----+---+-----+
最有效的方法是什么?
这不是人们能想出的最佳解决方案,但从我的头顶来看,它会完成这项工作:
import org.apache.spark.sql.Row
val df = sc.parallelize(List(("user1",1,5),("user2", 0, 13),("user2", 2, 4),("user3", 2, 7),("user3", 3, 45))).toDF("id","ind","freq")
df.show
// +-----+---+----+
// | id|ind|freq|
// +-----+---+----+
// |user1| 1| 5|
// |user2| 0| 13|
// |user2| 2| 4|
// |user3| 2| 7|
// |user3| 3| 45|
// +-----+---+----+
val df2 = df.groupBy('id).pivot("ind").max("freq").na.fill(0)
df2.show
// +-----+---+---+---+---+
// | id| 0| 1| 2| 3|
// +-----+---+---+---+---+
// |user1| 0| 5| 0| 0|
// |user2| 13| 0| 4| 0|
// |user3| 0| 0| 7| 45|
// +-----+---+---+---+---+
val cols = df2.columns
val df3 = df2.rdd.map {
case r : Row =>
val id = r.getAs[String]("id")
cols.map(ind => (id,ind,r.getAs[Integer](ind)))
}.flatMap(_.toSeq).filter(_._2 != "id").toDF("id","ind","freq")
df3.show
// +-----+---+----+
// | id|ind|freq|
// +-----+---+----+
// |user1| 0| 0|
// |user1| 1| 5|
// |user1| 2| 0|
// |user1| 3| 0|
// |user2| 0| 13|
// |user2| 1| 0|
// |user2| 2| 4|
// |user2| 3| 0|
// |user3| 0| 0|
// |user3| 1| 0|
// |user3| 2| 7|
// |user3| 3| 45|
// +-----+---+----+
我正在使用GroupeData
的pivot
函数,然后按列将其展平。(星火 1.6+(
PS:这个解决方案没有优化,我有很多缺点,即:大量的索引,计算成本等。
我刚刚遇到了这个"差距"问题,我想出的解决方案很幼稚,所以它可能效率不高,但我认为它非常简单。
基本上要穷尽(id, ind)
对的所有组合,从原始 DF 中选择不同的 id
值,然后再次选择不同的 ind
值,然后交叉连接这两个结果以获得所有组合。之后,只需将结果连接回原始DF,并用0填充所有NA。