为啥 Mongo Spark 连接器为查询返回不同且不正确的计数？

Posted 2023-04-15

【中文标题】为啥 Mongo Spark 连接器为查询返回不同且不正确的计数？

【英文标题】：Why Mongo Spark connector returns different and incorrect counts for a query?为什么 Mongo Spark 连接器为查询返回不同且不正确的计数？

【发布时间】：2017-10-09 17:35:38

【问题描述】：

我正在为一个项目评估 Mongo Spark 连接器，但得到的结果不一致。我在笔记本电脑上本地使用 MongoDB 服务器版本 3.4.5、Spark（通过 PySpark）版本 2.2.0、Mongo Spark 连接器版本 2.11;2.2.0。对于我的测试数据库，我使用安然数据集http://mongodb-enron-email.s3-website-us-east-1.amazonaws.com/ 我对 Spark SQL 查询很感兴趣，当我开始运行简单的计数测试查询时，每次运行都会收到不同的计数。 这是我的 mongo shell 的输出：

> db.messages.count('headers.To': '[e-mail]')
203

以下是我的 PySpark shell 的一些输出：

In [1]: df = spark.read.format("com.mongodb.spark.sql.DefaultSource").option("uri", "mongodb://127.0.0.1/enron_mail.messages").load()
In [2]: df.registerTempTable("messages")
In [3]: res = spark.sql("select count(*) from messages where headers.To='[e-mail]'")
In [4]: res.show()
+--------+                                                                      
|count(1)|
+--------+
|     162|
+--------+
In [5]: res.show()
+--------+                                                                      
|count(1)|
+--------+
|     160|
+--------+
In [6]: res = spark.sql("select count(_id) from messages where headers.To='[e-mail]'")
In [7]: res.show()
+----------+                                                                    
|count(_id)|
+----------+
|       161|
+----------+
In [8]: res.show()
+----------+                                                                    
|count(_id)|
+----------+
|       162|
+----------+

我在 Google 中搜索了有关此问题的信息，但没有发现任何有用的信息。如果有人对为什么会发生这种情况以及如何正确处理有任何想法，请分享您的想法。我有一种感觉，可能是我遗漏了某些内容，或者某些内容配置不正确。

更新： 我解决了我的问题。计数不一致的原因是 MongoDefaultPartitioner 包装了使用随机抽样的 MongoSamplePartitioner。老实说，这对我来说是一个非常奇怪的默认设置。我个人更喜欢使用缓慢但一致的分区器。分区器选项的详细信息可以在configuration options 官方文档中找到。

更新： 将解决方案复制到答案中。

【参考方案1】：

我解决了我的问题。计数不一致的原因是 MongoDefaultPartitioner 包装了使用随机抽样的 MongoSamplePartitioner。老实说，这对我来说是一个非常奇怪的默认设置。我个人更喜欢使用缓慢但一致的分区器。分区器选项的详细信息可以在configuration options 官方文档中找到。

代码：

val df = spark.read
  .format("com.mongodb.spark.sql.DefaultSource")
  .option("uri", "mongodb://127.0.0.1/enron_mail.messages")
  .option("partitioner", "spark.mongodb.input.partitionerOptions.MongoPaginateBySizePartitioner ")
  .load()

【讨论】：

@mrsrinivas 哦，对不起，我不知道。我之前检查了几个 scala 示例，所有示例都使用了 .load()。

【参考方案2】：

这个问题主要是由于 2.2.0 Mongo 连接器中的 SPARK-151 错误。它在 2.2.1 版本中得到解决，我已经确认。您可以继续使用 2.2.1 的默认分区器。