为什么Mongo Spark连接器会为查询返回不同和不正确的计数？

Question

我正在为一个项目评估Mongo Spark连接器,我得到了不一致的结果.我在我的笔记本电脑上使用MongoDB服务器版本3.4.5,Spark(通过PySpark)版本2.2.0,Mongo Spark Connector版本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': 'eric.bass@enron.com'})
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='eric.bass@enron.com'")
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='eric.bass@enron.com'")
In [7]: res.show()
+----------+                                                                    
|count(_id)|
+----------+
|       161|
+----------+
In [8]: res.show()
+----------+                                                                    
|count(_id)|
+----------+
|       162|
+----------+

我在谷歌搜索过这个问题,但我没有找到任何帮助.如果有人有任何想法,为什么会发生这种情况以及如何正确处理,请分享您的想法.我有一种感觉,也许我错过了一些东西,或者某些东西没有正确配置.

更新: 我解决了我的问题.计数不一致的原因是MongoDefaultPartitioner包含使用随机采样的MongoSamplePartitioner.说实话,对我来说,这是一个非常奇怪的默认.我个人更喜欢使用缓慢但一致的分区程序.有关分区程序选项的详细信息,请参阅官方配置选项文档.

更新: 将解决方案复制到答案中.

Answer 1

我解决了我的问题.计数不一致的原因是MongoDefaultPartitioner包含使用随机采样的MongoSamplePartitioner.说实话,对我来说,这是一个非常奇怪的默认.我个人更喜欢使用缓慢但一致的分区程序.有关分区程序选项的详细信息,请参阅官方配置选项文档.

码:

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()