Scala中DataFrame转JSON并存储为Mongo数组格式的问题

嘿，我之前也踩过这个坑！问题出在toJSON方法本身——它会把DataFrame的每一行转换成JSON字符串，而不是保留原生的结构化数据，所以MongoDB自然就把它们当成字符串类型存储了。下面给你几个靠谱的解决方案，按需选择：

这是最简洁高效的方式，不需要手动处理JSON转换，Spark Connector会帮你把DataFrame的结构映射成Mongo的文档结构。

步骤如下：

1. 确保你的项目已经引入了MongoDB Spark Connector依赖（比如Maven或SBT）。
2. 把原DataFrame转换成你需要的嵌套结构：先将每行的Word和Count字段包装成一个struct（重命名为word和count），再把所有struct收集成一个名为words的数组。
3. 直接写入MongoDB。

代码示例（Scala）：

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

// 假设你的WordCount结果DataFrame是wordCountDF
val targetDF = wordCountDF
  // 把每行的Word和Count转换成指定键名的struct
  .select(struct(col("Word").alias("word"), col("Count").alias("count")).alias("word_entry"))
  // 收集所有struct成数组，命名为words
  .agg(collect_list("word_entry").alias("words"))

// 配置Mongo连接信息并写入
targetDF.write
  .format("mongodb")
  .option("uri", "mongodb://localhost:27017/your_db.your_collection")
  .mode("append") // 根据需求选overwrite/ignore等
  .save()

这样写入Mongo的文档就是你想要的格式：{words: [{word:"abc",count:1}, {word:"xyz",count:23}]}

如果你的数据量很小，能完全加载到Driver内存里，可以先把DataFrame的数据转换成结构化对象，再用Mongo的官方Driver写入，避免字符串转换问题。

代码示例（Scala）：

// 定义匹配目标结构的Case Class
case class WordItem(word: String, count: Int)
case class WordResult(words: List[WordItem])

// 从DataFrame收集数据并转换成目标对象
val wordList = wordCountDF
  .select("Word", "Count")
  .as[(String, Int)] // 转换成Tuple类型
  .map { case (word, count) => WordItem(word, count) }
  .collect() // 收集到Driver端
  .toList

val finalResult = WordResult(wordList)

// 用Mongo Driver写入
import com.mongodb.client.MongoClients
import org.bson.codecs.configuration.CodecRegistries.{fromProviders, fromRegistries}
import org.mongodb.scala.bson.codecs.Macros._

// 注册Case Class的Codec，让Mongo能识别
val codecRegistry = fromRegistries(
  MongoClients.getDefaultCodecRegistry,
  fromProviders(classOf[WordItem], classOf[WordResult])
)

val client = MongoClients.create("mongodb://localhost:27017")
val db = client.getDatabase("your_db").withCodecRegistry(codecRegistry)
val coll = db.getCollection[WordResult]("your_collection")

// 插入文档
coll.insertOne(finalResult).subscribe()

⚠️ 注意：大数据量下不要用collect()，会把所有数据拉到Driver端，容易导致内存溢出，优先用方案1。

如果已经不小心把字符串存进去了，可以用MongoDB的聚合管道来批量转换：

db.your_collection.updateMany(
  {}, // 匹配所有文档
  [
    {
      $set: {
        words: { $jsonParse: "$words" } // 把字符串解析成JSON对象
      }
    }
  ]
)

这个命令会把所有文档中words字段的字符串内容转换成结构化的数组。

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