计算RDD中的行数

Question

我在Java中使用spark，并且具有500万行的RDD。有没有一种解决方案可以让我计算RDD的行数。我已经尝试过，RDD.count()但是要花很多时间。我已经知道我可以使用该功能fold。但是我没有找到此功能的Java文档。您能否请教我如何使用它，或给我另一个解决方案以获取RDD的行数。

这是我的代码：

JavaPairRDD<String, String> lines = getAllCustomers(sc).cache();
JavaPairRDD<String,String> CFIDNotNull = lines.filter(notNull()).cache();
JavaPairRDD<String, Tuple2<String, String>> join =lines.join(CFIDNotNull).cache();

double count_ctid = (double)join.count(); // i want to get the count of these three RDD
double all = (double)lines.count();
double count_cfid = all - CFIDNotNull.count();
System.out.println("********** :"+count_cfid*100/all +"% and now : "+ count_ctid*100/all+"%");

谢谢。

Answer 1

您有一个正确的想法：用于rdd.count()计算行数。没有更快的方法。

我认为您应该问的问题是 为什么rdd.count()这么慢？

答案是rdd.count()“动作”，这是一个急切的操作，因为它必须返回实际的数字。您之前执行的RDD操作count()是“转换”-他们将RDD延迟地转换为另一个。实际上，转换实际上并没有执行，只是排队。调用时count()，您将强制执行所有先前的惰性操作。输入的文件需要立即加载，执行map()s和filter()s，执行随机播放等，直到最后我们有了数据并可以说出它有多少行。

请注意，如果您拨打count()两次，所有这些都会发生两次。返回计数后，所有数据都将被丢弃！如果要避免这种情况，请调用cache()RDD。然后，对的第二次调用count()将很快，并且派生的RDD也会更快地计算出来。但是，在这种情况下，RDD必须存储在内存（或磁盘）中。