计算RDD中的行数

Question

我正在使用带有java的spark,我有一个500万行的RDD.是否有一个sollution,允许我计算我的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,执行shuffle等,直到最后我们有数据并且可以说它有多少行.

请注意,如果您拨打count()两次电话,所有这一切都会发生两次.返回计数后,所有数据都将被丢弃!如果您想避免这种情况,请致电cache()RDD.然后第二次调用count()将是快速的,并且派生的RDD将更快地计算.但是,在这种情况下,RDD必须存储在内存(或磁盘)中.

Answer 2

丹尼尔对这count笔钱的解释是正确的.但是,如果您愿意接受近似值,则可以尝试使用countApprox(timeout: Long, confidence: Double = 0.95): PartialResult[BoundedDouble]RDD方法.(注意,这被标记为"实验").