使用 pyspark 查询 MongoDB 集合

Question

我正在使用 PHP 和 MongoDB 作为数据库构建应用程序。跨 DB 的集合之一具有大量数据，即 8GB 数据。 我对存储在 MongoDB 集合中的数据执行 聚合 操作并相应地生成统计信息。 但是处理大量数据需要很长时间。因此我选择Apache spark来处理存储在MongDB集合中的数据 我已经配置了MongoDB spark connector 并在 python 中执行了一个演示脚本，以通过 spark 从 mongo 集合中获取数据。

以下是python代码sn -p

from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext
conf=SparkConf()
conf.set('spark.mongodb.input.uri','mongodb://[host]/db.collection')
conf.set('spark.mongodb.output.uri','mongodb://[host]/db.collection')
sc = SparkContext(conf=conf)
sqlContext = SQLContext(sc)
df = sqlContext.read.format("com.mongodb.spark.sql.DefaultSource").load()
df.printSchema()
df.registerTempTable("mycollection")
result_data=sqlContext.sql("SELECT * from mycollection limit 10")
result_data.show()

在上面的代码 sn-p 中，我使用了pyspark.sql module 来生成RDD。 但是 RDD 的生成会导致从集合中读取所有数据，这与 Apache Spark 的工作原理相反，需要很长时间才能读取大量数据。 因此建议我一个合适的解决方案来使用具有最佳性能的 pyspark 从 Mongo 集合中读取数据，以及是否存在任何 Apache spark 中的替代包 来与 MongoDB 通信强>.

Answer 1

不幸的是，与过滤器不同，limit 子句不会传递回 Mongo Spark 连接器，因此无法自动转换为聚合管道阶段。

但是，您始终可以提供自己的流水线阶段，例如：

dfr = sqlContext.read.format("com.mongodb.spark.sql.DefaultSource")
df = dfr.option("pipeline", "[{ $limit: 10 }]").load()