mongodb - 如何使用来自 MongoDB 的过滤记录构建 Spark 数据框？

Question

我的应用程序是使用 MongoDB 作为平台构建的。DB中的一个集合具有海量数据，并且选择了apache spark通过计算来检索和生成分析数据。我配置了Spark Connector for MongoDB与 MongoDB 通信。我需要使用 pyspark 查询 MongoDB 集合并构建一个由 mongodb 查询结果集组成的数据框。请给我一个合适的解决方案。

Answer 1

您可以像这样将数据直接加载到数据框中:

# Create the dataframe
df = sqlContext.read.format("com.mongodb.spark.sql.DefaultSource").option("uri", "mongodb://127.0.0.1/mydb.mycoll").load()

# Filter the data via the api
df.filter(people.age > 30)

# Filter via sql
df.registerTempTable("people")
over_thirty = sqlContext.sql("SELECT name, age FROM people WHERE age > 30")

有关详细信息，请参阅 Mongo Spark 连接器 Python API部分或 introduction.py . SQL 查询被翻译并传递回连接器，以便在将数据发送到 Spark 集群之前可以在 MongoDB 中查询数据。

您也可以提供自己的 aggregation pipeline在将结果返回到 Spark 之前应用到集合:

dfr = sqlContext.read.option("pipeline", "[{ $match: { name: { $exists: true } } }]")
df = dfr.option("uri", ...).format("com.mongodb.spark.sql.DefaultSource").load()