scala - 如何在 Spark 中对 RDD 进行排序和限制？

Question

我有 Foo 类的 RDD:class Foo( name : String, createDate : Date )。我想要另一个 RDD，其 Foo 旧 10%。我的第一个想法是按 createDate 排序并按 0.1*count 限制，但没有限制功能。

你有什么想法吗？

Answer 1

假设 Foo 是这样的案例类:

import java.sql.Date
case class Foo(name: String, createDate: java.sql.Date)

1. 使用普通 RDD:

   import org.apache.spark.rdd.RDD
   import scala.math.Ordering
   
   val rdd: RDD[Foo] = sc
     .parallelize(Seq(
       ("a", "2015-01-03"), ("b", "2014-11-04"), ("a", "2016-08-10"),
       ("a", "2013-11-11"), ("a", "2015-06-19"), ("a", "2009-11-23")))
     .toDF("name", "createDate")
     .withColumn("createDate", $"createDate".cast("date"))
     .as[Foo].rdd
   
   rdd.cache()
   val  n = scala.math.ceil(0.1 * rdd.count).toInt
   

   • 数据适合驱动程序内存:

     • 你想要的分数相对较小

       rdd.takeOrdered(n)(Ordering.by[Foo, Long](_.createDate.getTime))
       // Array[Foo] = Array(Foo(a,2009-11-23))
       

     • 您想要的分数相对较大:

       rdd.sortBy(_.createDate.getTime).take(n)
       

   • 否则

     rdd
       .sortBy(_.createDate.getTime)
       .zipWithIndex
       .filter{case (_, idx) => idx < n}
       .keys
     

2. 使用 DataFrame(注意 - 由于行为限制，这实际上不是最佳性能)。

   import org.apache.spark.sql.Row
   
   val topN = rdd.toDF.orderBy($"createDate").limit(n)
   topN.show
   
   // +----+----------+
   // |name|createDate|
   // +----+----------+
   // |   a|2009-11-23|
   // +----+----------+
   
   
   // Optionally recreate RDD[Foo]
   topN.map{case Row(name: String, date: Date) => Foo(name, date)}