pyspark操作MongoDB的方法步骤

下面是一份详细的“pyspark操作MongoDB的方法步骤”的攻略。

准备工作

在使用pyspark操作MongoDB前，请确保已经完成以下准备工作：

安装了pyspark和pymongo模块；
安装了MongoDB，并创建了需要操作的数据库及数据集合；
配置了MongoDB的用户名和密码，以保证连接MongoDB的权限。

步骤一：连接MongoDB数据库

pyspark操作MongoDB的第一步就是连接MongoDB数据库。在连接中需要指明MongoDB的地址、端口号、用户名和密码等相关信息。

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
import pymongo_spark

# 配置MongoDB的连接信息
mongodb_url = 'mongodb://<username>:<password>@<host>:<port>/<database>'

# 创建SparkSession
spark = SparkSession.builder.appName("mongo_spark").config("spark.mongodb.input.uri", mongodb_url)\
        .config("spark.mongodb.output.uri", mongodb_url).config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.12:3.0.1").getOrCreate()

# 加载MongoDB数据集
mongodb_df = spark.read.format("mongo").option("uri",mongodb_url).load()

步骤二：使用pyspark操作MongoDB数据库

连接MongoDB数据库成功后，我们就可以使用pyspark对数据库进行操作了，比如查询、新增、更新、删除等等。

示例一：查询MongoDB数据

下面是一个查询MongoDB数据的示例，代码如下：

# 查询MongoDB数据
result = mongodb_df.filter(col("age") > 30).select(col("name"), col("age")).show()

以上代码使用Spark DataFrame的方式，查询年龄大于30岁的数据，并只返回姓名和年龄两列。

示例二：新增MongoDB数据

下面是一个新增MongoDB数据的示例，代码如下：

# 新增MongoDB数据
new_data = {
    "name": "peter",
    "age": 25,
    "address": "Beijing"
}

mongodb = pymongo_spark.activate()
mongo_rdd = spark.sparkContext.parallelize([new_data])
mongodb.saveRDDToMongoDB(mongo_rdd)

以上代码使用Spark RDD的方式，新增一条数据到MongoDB数据库中。