scala – Spark：测试RDD是否为空的高效方法

根据this apache mail-thread及以后的建议，对此答案有一些意见，我做了一些小的本地实验。最好的方法是使用take(1)== 0。

def isEmpty[T](rdd : RDD[T]) = {
  rdd.take(1) == 0 
}

它应该在O(1)中运行，除非RDD为空，在这种情况下，分区数是线性的。

感谢Josh Rosen和Nick Chammas指出这一点。

注意：如果RDD类型为RDD [Nothing]，则失败。 isEmpty(sc.parallelize(Seq()))，但这在现实生活中可能不是问题。 isEmpty(sc.parallelize(Seq [Any]()))工作正常。

编辑：

>编辑1：添加take(1)== 0方法，感谢评论。

我原来的建议：使用mapPartitions。

def isEmpty[T](rdd : RDD[T]) = {
  rdd.mapPartitions(it => Iterator(!it.hasNext)).reduce(_&&_) 
}

它应该在分区数量上扩展，并不像以前那样干净(1)。然而RDD的类型是RDD [Nothing]。

实验：

我用这个代码来定时。

def time(n : Long,f : (RDD[Long]) => Boolean): Unit = {
  val start = System.currentTimeMillis()
  val rdd = sc.parallelize(1L to n,numSlices = 100)
  val result = f(rdd)
  printf("Time: " + (System.currentTimeMillis() - start) + "   Result: " + result)
}

time(1000000000L,rdd => rdd.take(1).length == 0L)
time(1000000000L,rdd => rdd.mapPartitions(it => Iterator(!it.hasNext)).reduce(_&&_))
time(1000000000L,rdd => rdd.count() == 0L)
time(1000000000L,rdd => rdd.takeSample(true,1).isEmpty)
time(1000000000L,rdd => rdd.fold(0)(_ + _) == 0L)

time(1L,rdd => rdd.take(1).length == 0L)
time(1L,rdd => rdd.mapPartitions(it => Iterator(!it.hasNext)).reduce(_&&_))
time(1L,rdd => rdd.count() == 0L)
time(1L,1).isEmpty)
time(1L,rdd => rdd.fold(0)(_ + _) == 0L)

time(0L,rdd => rdd.take(1).length == 0L)
time(0L,rdd => rdd.mapPartitions(it => Iterator(!it.hasNext)).reduce(_&&_))
time(0L,rdd => rdd.count() == 0L)
time(0L,1).isEmpty)
time(0L,rdd => rdd.fold(0)(_ + _) == 0L)

在我的本地机器上有3个工作核心，我得到了这些结果

Time:    21   Result: false
Time:    75   Result: false
Time:  8664   Result: false
Time: 18266   Result: false
Time: 23836   Result: false

Time:   113   Result: false
Time:   101   Result: false
Time:    68   Result: false
Time:   221   Result: false
Time:    46   Result: false

Time:    79   Result: true
Time:    93   Result: true
Time:    79   Result: true
Time:   100   Result: true
Time:    64   Result: true