升级MongoDB-Spark Connector至10.x后Map类型读为Struct的问题求助

MongoDB Spark Connector 10.x 系列对Spark Map类型的序列化/反序列化逻辑做了调整：

• 写入时，默认将Scala Map[String, Long] 序列化为MongoDB的内嵌BSON文档（即键作为字段名，值作为字段值）
• 读取时，这种内嵌BSON文档会被默认推断为Spark Struct类型，而非Map类型。即使开启sql.inferSchema.mapTypes.enabled=true也无效——该配置仅针对MongoDB中存储为键值对数组（如[{key: "k", value: v}, ...]）的数据，自动推断为Map类型。

旧版Connector 2.4.x的默认行为与新版不同，因此升级后出现类型不匹配的反序列化错误。同时，新版Connector移除了MongoSpark伴生对象，推荐使用标准Spark DataSource API（spark.read.format("mongodb")）替代。

方法一：修改写入配置，将Map序列化为键值对数组 #

在写入Dataset时添加spark.mongodb.output.mapMode=array配置，强制Spark Map类型序列化为MongoDB的键值对数组格式。这样读取时连接器会自动识别为Map类型，无需额外转换。

修改写入代码：

inputDS.write.format("mongodb")
  .mode("overwrite")
  .options(Map(
    "connection.uri" -> "MyConnectionURI",
    "database" -> "MyDatabaseName",
    "collection" -> "MyCollectionName",
    "replaceDocument" -> "false",
    "spark.mongodb.output.mapMode" -> "array" // 新增配置
  ))
  .save()

写入后，MongoDB中mapInfo字段的存储格式为：

"mapInfo": [
  {"key": "starfleet", "value": 10},
  {"key": "serenity", "value": 13}
]

此时读取的Schema会自动匹配原Map类型，直接执行outputDF.as[SimpleOutput]即可正常反序列化。

方法二：读取时显式将Struct转换为Map #

如果无法修改写入逻辑（比如已有存量数据为内嵌文档格式），可以通过Spark SQL函数将Struct类型的字段转换为Map类型：

import org.apache.spark.sql.functions._

val outputDF = spark.read.format("mongodb").options(readConfigOptions).load()

// 将Struct类型的mapInfo转换为Map[String, Long]
val convertedDF = outputDF.withColumn(
  "mapInfo",
  map_from_entries(
    array(
      // 遍历Struct的所有字段，生成键值对Struct数组
      outputDF.schema("mapInfo").fields.map(field =>
        struct(lit(field.name).alias("key"), col(s"mapInfo.${field.name}").alias("value"))
      ): _*
    )
  ).cast("map<string,long>")
)

// 转换为Dataset并验证
convertedDF.as[SimpleOutput].collect() should contain theSameElementsAs inputData

关于旧版MongoSpark.load[T]的替代 #

新版Connector推荐使用spark.read.format("mongodb").load().as[T]的方式直接将读取的DataFrame转换为Dataset，前提是读取的Schema与Case Class的字段类型完全匹配（方法一可满足此条件）。

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