Spark与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;
    at com.mongodb.spark.sql.connector.schema.InternalRowToRowFunction.<init>(InternalRowToRowFunction.java:44)
    at com.mongodb.spark.sql.connector.schema.RowToBsonDocumentConverter.<init>(RowToBsonDocumentConverter.java:84)
    at com.mongodb.spark.sql.connector.write.MongoDataWriter.<init>(MongoDataWriter.java:74)
    ...

# 假设 scala_version 和 spark_version 已定义
scala_version = "2.12" # 示例值，请根据实际环境设置
spark_version = "3.5.0" # 示例值，请根据实际环境设置

# ----------------------------------------------------------
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.1 (或更高版本，以兼容Spark 3.5.0)
    "org.mongodb.spark:mongo-spark-connector_2.12:10.3.1" 
]

spark = SparkSession.builder \
    .master(MASTER) \
    .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")

# 示例写入函数（保持不变，因为问题出在依赖版本）
def write_to_db(df, epoch_id):
    df.select("id", "production_companies").show()
    df.write.format("mongodb") \
               .mode("append") \
               .option("database", "BIGDATA") \
               .option("collection", "movie") \
               .save()

# 假设您有一个流式DataFrame 'streaming_df'
# streaming_df.writeStream \
#     .foreachBatch(write_to_db) \
#     .outputMode("update") \
#     .start() \
#     .awaitTermination()