сортировка и ранжирование в искровой RDD в одном файле - PullRequest
Новая
       6

сортировка и ранжирование в искровой RDD в одном файле

0 голосов
/ 20 ноября 2018

У меня есть искровой СДР, как показано ниже 

(maths,60)
(english,65)
(english,77)
(maths,23)
(maths,50)

Мне нужно отсортировать и ранжировать данный СДР в один, как показано ниже (математика, 23,1) (математика, 50,2) (математика, 50, 3) (английский, 65,1) (английский, 77,2)

я знаю, что это легко сделать с помощью Data Frame, но мне нужен Spark rdd код, чтобы получить решение, пожалуйста, предложите

Ответы [ 2 ]

0 голосов
/ 20 ноября 2018

Другое странное решение: 

val df = Seq(("maths",60),("english",65),("english",77),("maths",23),("maths",50)).toDF("subject","marks")
val rdd1 = df.rdd
rdd1.groupBy( x=> x(0))
  .map( x=> 
      {
         val p = x._2.toList.map(a=>a(1)).map(_.toString.toInt).sortWith((a1,a2)=> a1 < a2 ).zipWithIndex.map(b=>(b._1,b._2+1))
        (x._1,p) 
      }
  )
  .flatMap( x => x._2.map((x._1,_)))
  .collect.foreach(println)

Результаты: 

(english,(65,1))
(english,(77,2))
(maths,(23,1))
(maths,(50,2))
(maths,(60,3))
0 голосов
/ 20 ноября 2018

Spark RDD функции (так называемые преобразования), такие как groupByKey flatMap и Scala List функция, такая как sorted, должна помочь в ее достижении. 

val rdd = spark.sparkContext.parallelize(
  Seq(("maths",60), 
      ("english",65), 
      ("english",77), 
      ("maths",23), 
      ("maths",50)))

val result = rdd.groupByKey().flatMap(group => {

  group._2.toList
  .sorted.toList // sort marks
  .zipWithIndex // add the position/rank
  .map {

    case(marks, index) => (group._1, marks, index + 1)
  }
})

result.collect

// Array((english,65,1), (english,77,2), (maths,23,1), (maths,50,2), (maths,60,3))

Блокнот данных блокнот

...