PySpark：无法找到数据源 com.mongodb.spark.sql.DefaultSource

在本文中，我们将介绍在 PySpark 中遇到的一种常见错误，即“Failed to find data source: com.mongodb.spark.sql.DefaultSource”。我们将探讨该错误的原因和解决方法，并提供示例说明。

阅读更多：PySpark 教程

问题描述

在使用 PySpark 进行数据处理时，我们经常需要从外部数据源中读取和写入数据。其中，MongoDB 是一种常见的 NoSQL 数据库，PySpark 通过 spark-mongodb 包提供了与 MongoDB 的集成支持。然而，在使用 PySpark 读取或写入 MongoDB 数据时，可能会遇到以下错误信息：

pyspark.sql.utils.ParseException: Failed to find data source: com.mongodb.spark.sql.DefaultSource

错误原因

这个错误通常是由于缺少 spark-mongodb 包导致的。PySpark 默认并不内置对 MongoDB 的支持，需要手动添加相应的包才能使用与 MongoDB 相关的功能。

解决方法

解决此问题的方法是安装并导入 spark-mongodb 包。以下是在 PySpark 中解决此错误的步骤：

步骤 1：添加 Maven 依赖

首先，我们需要在 PySpark 中添加 spark-mongodb 包的 Maven 依赖。在 PySpark 代码的开头部分，通过 SparkSession.builder 添加 Maven 依赖，使用 --packages 参数指定要导入的包及其版本号。例如：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("MongoDB Integration") \
    .config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.12:3.0.1") \
    .getOrCreate()

上述代码中，我们指定了 org.mongodb.spark:mongo-spark-connector_2.12:3.0.1 版本的 spark-mongodb 包。

步骤 2：配置 MongoDB 连接

接下来，我们需要在代码中配置 MongoDB 连接。在上述代码的基础上，添加如下部分：

spark.conf.set("spark.mongodb.input.uri", "mongodb://<hostname>:<port>/<database>.<collection>")
spark.conf.set("spark.mongodb.output.uri", "mongodb://<hostname>:<port>/<database>.<collection>")

上述代码中，需要将 <hostname>、<port>、<database> 和 <collection> 替换为实际的 MongoDB 连接信息。这里是示例，实际使用时需要根据 MongoDB 的配置进行调整。

步骤 3：读取和写入 MongoDB 数据

现在，我们已经成功导入了 spark-mongodb 包，并配置了 MongoDB 连接，可以使用 PySpark 读取和写入 MongoDB 数据了。以下是一些示例代码：

读取 MongoDB 数据

df = spark.read.format("com.mongodb.spark.sql.DefaultSource").load()
df.show()

写入数据到 MongoDB

data = [("Alice", 25), ("Bob", 30), ("Charlie", 35)]
df = spark.createDataFrame(data, ["name", "age"])
df.write.format("com.mongodb.spark.sql.DefaultSource").mode("overwrite").save()

总结

本文介绍了在使用 PySpark 读取或写入 MongoDB 数据时，可能遇到的错误“Failed to find data source: com.mongodb.spark.sql.DefaultSource”。我们解释了该错误的原因是缺少 spark-mongodb 包，并提供了解决方法。通过添加 Maven 依赖和配置 MongoDB 连接，我们可以成功地读取和写入 MongoDB 数据。

希望本文可以帮助你解决在 PySpark 中遇到的该问题。为了避免这类错误，记得在开始使用 PySpark 与 MongoDB 进行数据处理时导入相应的依赖包。