从MongoDB读取

您可以创建一个Spark DataFrame来保存来自您正在使用的spark.mongodb.input.uri选项中 指定的MongoDB集合的数据 SparkSession。

考虑一个名为的集合fruit，其中包含以下文档：

{ "_id" : 1, "type" : "apple", "qty" : 5 }
{ "_id" : 2, "type" : "orange", "qty" : 10 }
{ "_id" : 3, "type" : "banana", "qty" : 15 }

spark.read() 从pysparkshell 内将集合分配给DataFrame 。

df = spark.read.format("mongo").load()

Spark对记录进行采样以推断集合的模式。

df.printSchema()

上面的操作产生以下shell输出：

root
 |-- _id: double (nullable = true)
 |-- qty: double (nullable = true)
 |-- type: string (nullable = true)

如果需要从其他MongoDB集合中读取数据，则在将数据读取到DataFrame中时使用.option方法。

要从名为contacts的数据库中 的集合中读取内容people，请people.contacts在input URI选项中指定。

df = spark.read.format("mongo").option("uri",
"mongodb://127.0.0.1/people.contacts").load()