相关疑难解决方法(0)

我正在使用带有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+"%");

谢谢.

使用spark保存记录时,有没有办法获取写入的记录数？虽然我知道目前不符合规范,但我希望能够做到这样的事情:

val count = df.write.csv(path)

或者,能够进行步骤结果的内联计数(优选地不使用标准累加器)将(几乎)同样有效.即:

dataset.countTo(count_var).filter({function}).countTo(filtered_count_var).collect()

有任何想法吗？

考虑在spark中有2个表或表引用要比较,例如,以确保备份正常工作.是否有可能在火花中做那个遥控？因为使用将所有数据复制到R没有用collect().

library(sparklyr)
library(dplyr)
library(DBI)

##### create spark connection here
# sc <- spark_connect(<yourcodehere>)
spark_connection(sc)
spark_context(sc)

trees1_tbl <- sdf_copy_to(sc, trees, "trees1")
trees2_tbl <- sdf_copy_to(sc, trees, "trees2")
identical(trees1_tbl, trees2_tbl) # FALSE
identical(collect(trees1_tbl), collect(trees2_tbl)) # TRUE
setequal(trees1_tbl, trees2_tbl) # FALSE
setequal(collect(trees1_tbl), (trees2_tbl)) # TRUE

spark_disconnect(sc)

会很好,如果dplyr::setequal()可以直接使用.

通常我Dataset.count在 3 个场景中看到整个代码库：

1. 日志记录 log.info("this ds has ${dataset.count} rows")
2. 分枝 if (dataset.count > 0) do x else do y
3. 强制缓存 dataset.persist.count

它是否会通过强制查询优化器在任何这些场景中过早地急切来阻止查询优化器创建最有效的 dag？