将 csv 文件与 pySpark 进行比较答案

【问题标题】：Comparing csv files with pySpark将 csv 文件与 pySpark 进行比较
【发布时间】：2016-10-18 00:15:19
【问题描述】：

我是 pyspark 的新手，但我需要快速深入了解它。我想比较 pyspark 中的两个（巨大的）csv 文件并且到目前为止管理得很好（我很确定，我的代码并不花哨）最后我想计算匹配的记录和不匹配的记录。

我能做到的是：

1. 将 csv 加载到 RDD 中。

act="actual.csv"
exp="expected.csv"
raw_exp = sc.textFile(exp)                                                  
raw_act = sc.textFile(act)

2.我可以使用.count()计算记录的数量

print "Expected: ", raw_exp.count()
print "Actual:", raw_act.count()

3.我可以通过使用减法和收集来比较rdds以获得不匹配的记录：

notCompRecords  = raw_exp.subtract(raw_act).collect()

现在我想计算那些不匹配的记录。我想我会用：

notCompRecords.count()

但我得到一个参数丢失的错误：

TypeError: count() takes at least 1 argument (0 given)

我还了解到，我必须通过以下方式将 notComRecords 明显为字符串的列表转换为字符串：

notCompString   = ''.join(notCompRecords)

但这也行不通。

如何计算 Object/Variable/rdd notCompRecords 中的行数？

谢谢！任何提示或线索表示赞赏。最好的问候，

【问题讨论】：

从notCompRecords = raw_exp.subtract(raw_act).collect() 中删除.collect()。之后你可以使用notCompRecords.count()。
哇！进展顺利，非常快！非常感谢！
感谢您给予快速保证这有效，有时人们在他们的问题得到解决时不会回来。现在我们可能会支持@jho 的评论，因为它确实为问题添加了一些东西 - 一个答案;-)
刚刚添加它作为答案。

【解决方案1】：

从 notCompRecords = raw_exp.subtract(raw_act).collect() 中删除 .collect()。之后你可以使用 notCompRecords.count()。

【讨论】：