sparkscala：将dataframe列值聚合到有序列表中

q3aa0525 于 2021-05-27 发布在 Spark

关注(0)|答案(2)|浏览(427)

我有一个sparkscalaDataframe，它有四个值：（id，day，val，order）。我想创建一个新的dataframe，包含以下列：（id，day，value_list:list（val1，val2，…，valn）），其中val1到valn按asc order value排序。
例如：

(50, 113, 1, 1), 
(50, 113, 1, 3), 
(50, 113, 2, 2), 
(51, 114, 1, 2), 
(51, 114, 2, 1), 
(51, 113, 1, 1)

将变成：

((51,113),List(1))
((51,114),List(2, 1)
((50,113),List(1, 2, 1))

我很接近，但不知道在我把数据汇总成一个列表后该怎么办。我不知道如何按int的顺序排列每个值列表：

import org.apache.spark.sql.Row

val testList = List((50, 113, 1, 1), (50, 113, 1, 3), (50, 113, 2, 2), (51, 114, 1, 2), (51, 114, 2, 1), (51, 113, 1, 1))
val testDF = sqlContext.sparkContext.parallelize(testList).toDF("id1", "id2", "val", "order")

val rDD1 = testDF.map{case Row(key1: Int, key2: Int, val1: Int, val2: Int)  => ((key1, key2), List((val1, val2)))}
val rDD2 = rDD1.reduceByKey{case (x, y) =>  x ++ y}

输出如下所示：

((51,113),List((1,1)))
((51,114),List((1,2), (2,1)))
((50,113),List((1,3), (1,1), (2,2)))

下一步是生产：

((51,113),List((1,1)))
((51,114),List((2,1), (1,2)))
((50,113),List((1,1), (2,2), (1,3)))

scala apache-spark

来源：https://stackoverflow.com/questions/40295107/spark-scala-aggregate-dataframe-column-values-into-a-ordered-list

2条答案

按热度按时间

n53p2ov01#

你只需要在Map上标出你的 RDD 使用 sortBy :

scala> val df = Seq((50, 113, 1, 1), (50, 113, 1, 3), (50, 113, 2, 2), (51, 114, 1, 2), (51, 114, 2, 1), (51, 113, 1, 1)).toDF("id1", "id2", "val", "order")
df: org.apache.spark.sql.DataFrame = [id1: int, id2: int, val: int, order: int]

scala> import org.apache.spark.sql.Row
import org.apache.spark.sql.Row

scala> val rDD1 = df.map{case Row(key1: Int, key2: Int, val1: Int, val2: Int)  => ((key1, key2), List((val1, val2)))}
rDD1: org.apache.spark.rdd.RDD[((Int, Int), List[(Int, Int)])] = MapPartitionsRDD[10] at map at <console>:28

scala> val rDD2 = rDD1.reduceByKey{case (x, y) =>  x ++ y}
rDD2: org.apache.spark.rdd.RDD[((Int, Int), List[(Int, Int)])] = ShuffledRDD[11] at reduceByKey at <console>:30

scala> val rDD3 = rDD2.map(x => (x._1, x._2.sortBy(_._2)))
rDD3: org.apache.spark.rdd.RDD[((Int, Int), List[(Int, Int)])] = MapPartitionsRDD[12] at map at <console>:32

scala> rDD3.collect.foreach(println)
((51,113),List((1,1)))
((50,113),List((1,1), (2,2), (1,3)))
((51,114),List((2,1), (1,2)))

赞(0）回复(0）举报 2021-05-27

d8tt03nd2#

testDF.groupBy("id1","id2").agg(collect_list($"val")).show
+---+---+-----------------+                                                     
|id1|id2|collect_list(val)|
+---+---+-----------------+
| 51|113|              [1]|
| 51|114|           [1, 2]|
| 50|113|        [1, 1, 2]|
+---+---+-----------------+

赞(0）回复(0）举报 2021-05-27

我来回答

sparkscala：将dataframe列值聚合到有序列表中

2条答案

相关问题

热门标签

最新问答