스파크 스칼라 : DataFrame 열 값을 정렬 된 목록으로 집계

Question

저는 (id, day, val, order)의 4 가지 값을 가진 스파크 스칼라 DataFrame을 가지고 있습니다. (id, day, value_list : List (val1, val2, ..., valn)) 여기서 val1, valn은 오름차순으로 정렬됩니다. 예를 들어

:

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

나는 가까운,하지만 난이 목록에 데이터를 집계 한 후 어떻게해야할지 모르겠어요. 출력과 같은

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

다음 단계는 생산하는 것입니다 :

그때 주문 INT하여 각 값 목록 점화 순서가하는 방법을 모르겠어요

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

Answer 1

당신은 당신의 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))) 

Answer 2

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