使用MongoDB Spark Connector 10.2.1流式处理数据报错求助

问题原因 #

你遇到的java.lang.NoSuchMethodError: org.apache.spark.sql.types.StructType.toAttributes()错误，本质是版本兼容性问题：

• MongoDB Spark Connector 10.2.1要求依赖Spark 3.3.x及以上版本，该版本引入了StructType.toAttributes()方法
• 你当前使用的PySpark是3.2.3，这个版本中不存在该方法，因此触发报错。

解决方案 #

提供两种可行方案，根据你的环境选择：

方案1：升级PySpark到兼容版本 #

将PySpark版本升级到3.3.x或更高（比如3.3.4、3.4.x），保持MongoDB Spark Connector 10.2.1不变。升级后重新运行代码即可解决版本适配问题。

方案2：降级MongoDB Spark Connector到适配Spark 3.2的版本 #

如果你无法升级PySpark，可将Connector版本降级到10.1.1（该版本官方明确支持Spark 3.2.x + Scala 2.13），修改SparkSession配置中的spark.jars.packages：

.config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.13:10.1.1")

代码修正（匹配你的需求） #

注意到你的需求是流式读取MongoDB数据并打印到控制台，但当前代码是将数据写入到另一个MongoDB集合contacts2，需修改writeStream部分：

my_spark = SparkSession \
.builder \
.appName("myApp") \
.config("spark.mongodb.input.uri", "mongodb://mongoadmin:mongoadmin@localhost:27018/") \
# 若选择方案2，这里修改Connector版本
.config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.13:10.1.1") \
.getOrCreate()

# 流式读取MongoDB数据
streamingDataFrame = (my_spark.readStream
                  .format("mongodb")
                  .option("database", "people")
                  .option("collection", "contacts")
                  .load()
                  )

# 修改为控制台输出，匹配你的需求
dataStreamWriter = (streamingDataFrame.writeStream
                .trigger(processingTime="1 second")  # 连续模式对控制台输出支持有限，推荐用processingTime触发
                .format("console")
                .outputMode("append")
                )

query = dataStreamWriter.start()
query.awaitTermination()  # 添加该方法保持流持续运行

注：连续模式（continuous="1 second"）主要面向低延迟场景，对控制台输出的支持不完善，建议改用processingTime触发模式。

内容的提问来源于stack exchange，提问作者amirhossein