解决Spark 3.5.0与MongoDB连接器版本不兼容导致的写入错误

本文详细探讨了在使用spark 3.5.0通过`mongo-spark-connector`写入mongodb时遇到的`nosuchmethoderror`。该错误通常源于连接器与spark版本不兼容。核心解决方案是升级`mongo-spark-connector`至10.3.x版本，以确保其与spark 3.1至3.5.0的兼容性，从而实现数据的顺利写入。

Spark与MongoDB数据写入中的版本兼容性挑战

在大数据处理领域，Spark与各种数据存储系统（如MongoDB）的集成是常见的操作。然而，由于Spark生态系统的快速发展和第三方连接器库的独立维护，版本兼容性问题常常成为开发者面临的挑战。一个典型的表现就是运行时抛出的java.lang.NoSuchMethodError。这种错误通常意味着程序尝试调用一个在当前JVM环境中不存在的方法，或者该方法的签名与编译时所依赖的版本不签名。在Spark与MongoDB的集成场景中，这通常指向mongo-spark-connector与当前Spark版本之间的不匹配。

NoSuchMethodError的深入分析：Spark SQL内部API变更

当Spark应用程序在尝试将DataFrame写入MongoDB时遇到如下错误：

java.lang.NoSuchMethodError: org.apache.spark.sql.catalyst.encoders.RowEncoder$.apply(Lorg/apache/spark/sql/types/StructType;)Lorg/apache/spark/sql/catalyst/encoders/ExpressionEncoder;

这个错误信息揭示了问题的核心。org.apache.spark.sql.catalyst.encoders.RowEncoder是Spark SQL内部用于将Row对象编码（序列化）为内部格式或解码（反序列化）的关键组件。catalyst包下的类通常是Spark内部实现细节，不作为公共API暴露。这意味着mongo-spark-connector在处理Spark DataFrame时，依赖了Spark内部的RowEncoder的一个特定方法签名。

NoSuchMethodError的出现表明，当前运行的Spark版本（例如Spark 3.5.0）中，RowEncoder$对象（Scala伴生对象）的apply方法签名与mongo-spark-connector 10.2.1版本编译时所预期的签名不一致。Spark的不同版本，尤其是在主要或次要版本更新时，其内部API可能会发生变化，导致依赖旧版API的第三方库在新版Spark上运行时出现二进制兼容性问题。因此，问题根源在于mongo-spark-connector 10.2.1版本未能完全兼容Spark 3.5.0的内部API。

解决方案：升级MongoDB Spark Connector版本

解决此类NoSuchMethodError最直接有效的方法是更新导致兼容性问题的库版本。针对Spark 3.5.0与MongoDB的集成，mongo-spark-connector的官方文档明确指出，版本10.3.x提供了对Spark 3.1到3.5.0的全面支持。因此，将项目中使用的mongo-spark-connector从10.2.1升级到10.3.x（例如10.3.1或更高兼容版本）是解决此问题的关键。

挖错网

一款支持文本、图片、视频纠错和AIGC检测的内容审核校对平台。

28

查看详情

示例代码：更新依赖配置

以下是更新SparkSession配置中spark.jars.packages以使用兼容的mongo-spark-connector版本的示例代码：

from pyspark.sql import SparkSession
from pyspark.sql.streaming import StreamingQuery

# 假设 scala_version 和 spark_version 已在环境中定义
scala_version = "2.12" # 确保与Spark编译的Scala版本一致
spark_version = "3.5.0" # 目标Spark版本

# 定义所有需要的JAR包
packages = [
    f'org.apache.spark:spark-sql-kafka-0-10_{scala_version}:{spark_version}',
    'org.apache.kafka:kafka-clients:3.5.0',
    'org.apache.hadoop:hadoop-client:3.0.0',
    'org.elasticsearch:elasticsearch-spark-30_2.12:7.17.16',
    # 关键修改：将 mongo-spark-connector 版本升级到 10.3.x
    "org.mongodb.spark:mongo-spark-connector_2.12:10.3.1" # 推荐使用10.3.1或更高兼容版本
]

# 初始化SparkSession
spark = SparkSession.builder \
    .master("local[*]") \
    .appName("Movie Consumer") \
    .config("spark.jars.packages", ",".join(packages)) \
    .config(f"spark.mongodb.input.uri", f"mongodb+srv://<USERNAME>:<PASSWORD>@atlascluster.zdoemtz.mongodb.net") \
    .config(f"spark.mongodb.output.uri", f"mongodb+srv://<USERNAME>:<PASSWORD>@atlascluster.zdoemtz.mongodb.net") \
    .config("spark.cores.max", "1") \
    .config("spark.executor.memory", "1g") \
    .getOrCreate()

spark.sparkContext.setLogLevel("ERROR")

# 定义写入MongoDB的函数
def write_to_db(df, epoch_id):
    """
    将DataFrame数据写入MongoDB。
    :param df: 要写入的DataFrame
    :param epoch_id: Spark Streaming批次ID
    """
    print(f"Processing epoch {epoch_id}, writing {df.count()} rows to MongoDB.")
    df.write.format("mongodb") \
               .mode("append") \
               .option("database", "BIGDATA") \
               .option("collection", "movie") \
               .save()

# 示例：假设这是一个Spark Streaming的foreachBatch调用
# stream_df = spark \
#     .readStream \
#     .format("kafka") \
#     .option("kafka.bootstrap.servers", "localhost:9092") \
#     .option("subscribe", "movie_topic") \
#     .load()
#
# query: StreamingQuery = stream_df.writeStream \
#     .foreachBatch(write_to_db) \
#     .outputMode("update") \
#     .start()
#
# query.awaitTermination()

# 在实际应用中，确保替换 <USERNAME> 和 <PASSWORD> 为您的MongoDB Atlas凭据。
# 如果不是使用MongoDB Atlas，请相应调整URI。

通过将org.mongodb.spark:mongo-spark-connector_2.12:10.2.1替换为org.mongodb.spark:mongo-spark-connector_2.12:10.3.1（或根据MongoDB Spark Connector官方文档推荐的最新兼容版本），即可解决因版本不兼容导致的NoSuchMethodError。

注意事项与最佳实践

1. 查阅官方文档： 在集成任何第三方库时，务必首先查阅其官方文档，特别是关于版本兼容性矩阵的部分。这是解决版本冲突最权威的来源。
2. Scala版本匹配： Spark、连接器以及其他相关库通常会针对特定的Scala版本进行编译（例如_2.12）。确保项目中所有组件的Scala版本保持一致，以避免进一步的二进制兼容性问题。
3. 依赖管理： 谨慎管理spark.jars.packages或--packages参数。只引入必需的依赖，并尽量使用官方推荐的版本组合。过度或不当的依赖引入可能导致Jar包冲突（Jar Hell）。
4. 错误日志分析： 遇到NoSuchMethodError或ClassNotFoundException等运行时错误时，应首先检查依赖库的版本是否与运行环境兼容。详细的错误堆栈信息是诊断问题的关键线索。
5. 测试环境先行： 在生产环境部署前，务必在与生产环境尽可能一致的测试环境中验证所有组件的版本兼容性。

总结

java.lang.NoSuchMethodError在Spark与第三方库集成中是一个常见的版本兼容性问题。当Spark 3.5.0与mongo-spark-connector 10.2.1结合使用时，由于Spark内部API的变更，导致了此错误。核心解决方案是将mongo-spark-connector升级到官方推荐的10.3.x版本，该版本提供了对Spark 3.1至3.5.0的兼容性。通过遵循官方文档、细致的依赖管理和严格的测试，可以有效避免此类问题，确保Spark应用程序的稳定高效运行。

以上就是解决Spark 3.5.0与MongoDB连接器版本不兼容导致的写入错误的详细内容，更多请关注php中文网其它相关文章！