pyspark 針對mongodb的讀寫

1.創建pyspark與mongodb的連接，首先加載依賴包，其有三種方式：

1）直接將其放在在安裝spark的jars目錄下；

2）在spark_submit中，添加依賴包信息；

3）在創建spark的對象的時候添加依賴信息，具體案例如下圖所示

spark = SparkSession \
    .builder \
    .appName('mongo connection') \
    .config("spark.mongodb.input.uri", "mongodb://節點:端口號/dev.myCollection?readPreference=primaryPreferred") \
    .config("spark.mongodb.output.uri", "mongodb://節點:端口號/dev.myCollection") \
    .config('spark.jars.packages', "org.mongodb.spark:mongo-spark-connector_2.11:2.4.2") \
    .getOrCreate()

備注：

config的信息，都可以在spark_submit中添加。

2.讀取mongodb

df1 = (
spark.read
.format("mongo")
.option("database", 'dev')
.option("collection", 'test_mongo_connect')
.load()
)

3.寫入mongodb

df = spark.createDataFrame([(1,), (2,)], ['a'])
(
df.write
.format("mongo")
.mode("overwrite")
.option("database", 'dev')
.option("collection", 'test_mongo_connect')
.save()
)