从MongoDB到Pyspark DataFrame

要将MongoDB中的数据导入到Pyspark DataFrame中，可以按照以下步骤进行：

1. 安装所需的库：确保已安装pymongo和pyspark库。可以使用以下命令进行安装：

   pip install pymongo
   pip install pyspark
   

2. 导入所需的库和模块：

   import pymongo
   from pyspark.sql import SparkSession
   

3. 创建MongoDB连接并获取数据：

   # 创建MongoDB连接
   client = pymongo.MongoClient("<mongo_connection_string>")
   
   # 选择数据库和集合
   db = client["<database_name>"]
   collection = db["<collection_name>"]
   
   # 获取数据
   data = collection.find()
   

4. 创建SparkSession对象：

   # 创建SparkSession对象
   spark = SparkSession.builder \
       .appName("MongoDB to PySpark DataFrame") \
       .getOrCreate()
   

5. 将MongoDB数据转换为Pyspark DataFrame：

   # 将MongoDB数据转换为Pyspark DataFrame
   df = spark.createDataFrame(data)
   

6. 可选：进行数据处理和转换：

   # 进行数据处理和转换
   df = df.select("column1", "column2")  # 选择需要的列
   df = df.filter(df.column1 > 10)  # 过滤数据
   

7. 显示DataFrame的内容：

   # 显示DataFrame的内容
   df.show()
   

完整的代码示例：

import pymongo
from pyspark.sql import SparkSession

# 创建MongoDB连接
client = pymongo.MongoClient("<mongo_connection_string>")

# 选择数据库和集合
db = client["<database_name>"]
collection = db["<collection_name>"]

# 获取数据
data = collection.find()

# 创建SparkSession对象
spark = SparkSession.builder \
    .appName("MongoDB to PySpark DataFrame") \
    .getOrCreate()

# 将MongoDB数据转换为Pyspark DataFrame
df = spark.createDataFrame(data)

# 进行数据处理和转换
df = df.select("column1", "column2")  # 选择需要的列
df = df.filter(df.column1 > 10)  # 过滤数据

# 显示DataFrame的内容
df.show()

请注意，上述代码中的<mongo_connection_string>，<database_name>和<collection_name>应该替换为实际的MongoDB连接字符串、数据库名称和集合名称。