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

q3aa0525  于 2021-05-27  发布在  Spark
关注(0)|答案(2)|浏览(426)

我有一个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)))
n53p2ov0

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)))
d8tt03nd

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]|
+---+---+-----------------+

相关问题