Spark Connector Python指南

Python Spark Shell

本教程使用pyspark外壳程序，但是代码也可用于独立的Python应用程序。

启动pyspark外壳程序时，可以指定：

• --packages下载MongoDB Spark Connector软件包的选项。提供以下软件包：

  • mongo-spark-connector_2.11 用于Scala 2.11.x

• --conf用于配置MongoDB Spark Connnector 的选项。这些设置配置SparkConf对象。

  注意

  通过指定连接器配置时SparkConf，您必须在设置之前添加适当的前缀。有关详细信息和其他可用的MongoDB Spark Connector选项，请参阅“ 配置选项”。

以下示例pyspark从命令行启动外壳程序：

./bin/pyspark --conf "spark.mongodb.input.uri=mongodb://127.0.0.1/test.myCollection?readPreference=primaryPreferred" \
              --conf "spark.mongodb.output.uri=mongodb://127.0.0.1/test.myCollection" \
              --packages org.mongodb.spark:mongo-spark-connector_2.11:2.4.1

• 该spark.mongodb.input.uri指定MongoDB服务器地址（127.0.0.1），数据库连接（test），和收集（myCollection）从中读取数据，读偏好。
• 该spark.mongodb.output.uri指定MongoDB服务器地址（127.0.0.1），数据库连接（test），和收集（myCollection）要写入的数据。27017默认情况下连接到端口。
• 该packages选项以格式指定Spark Connector的Maven坐标groupId:artifactId:version。

本教程中的示例将使用此数据库和集合。

创建一个SparkSession对象

如果在启动时指定spark.mongodb.input.uri 和spark.mongodb.output.uri配置选项pyspark，则默认SparkSession对象将使用它们。如果您想SparkSession从中创建自己的对象 pyspark，则可以使用SparkSession.builder并指定不同的配置选项。

from pyspark.sql import SparkSession

my_spark = SparkSession \
    .builder \
    .appName("myApp") \
    .config("spark.mongodb.input.uri", "mongodb://127.0.0.1/test.coll") \
    .config("spark.mongodb.output.uri", "mongodb://127.0.0.1/test.coll") \
    .getOrCreate()

您可以使用SparkSession对象将数据写入MongoDB，从MongoDB读取数据，创建DataFrame以及执行SQL操作。