MongoSpark技术问询：BSON Document转Map[String,Double]及嵌套文档入DataFrame

嘿，我来帮你搞定这个难题！你提到的Spark生成的SparseVector Map形式的嵌套文档，确实没法直接被Spark自动解析成MLlib能用的SparseVector类型，不过我们可以通过自定义解析逻辑来解决这个问题，步骤很清晰：

第一步：先读入原始MongoDB数据 #

首先确保你已经配置好了MongoDB Spark Connector（根据你的Spark版本选对应的Connector版本就行），然后先把数据读成原始DataFrame——这时候嵌套的SparseVector会被当成普通的Map或者Struct类型：

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("MongoDBSparseVectorParser")
  .config("spark.mongodb.read.connection.uri", "mongodb://localhost:27017/你的数据库名.你的集合名")
  .getOrCreate()

// 读取原始数据，此时Genres这类字段是Map/Struct类型
val rawDF = spark.read.format("mongodb").load()
rawDF.printSchema() // 可以先看看原始结构

第二步：根据你的SparseVector格式写自定义UDF #

Spark的SparseVector需要三个核心参数：总维度（size）、非零元素的索引数组（indices）、对应的值数组（values）。我们要根据你MongoDB里的Map格式，写一个UDF把它转成标准的SparseVector。

场景1：Map里包含size、indices、values三个字段 #

如果你的嵌套文档（比如Genres）是这种结构：

{"size": 10, "indices": [0,2], "values": [1.0, 0.8]}

那UDF可以这么写：

import org.apache.spark.ml.linalg.SparseVector
import org.apache.spark.sql.functions.{udf, col}

// 定义UDF：从Struct/Map提取三个参数生成SparseVector
val structToSparse = udf { (size: Int, indices: Seq[Int], values: Seq[Double]) =>
  new SparseVector(size, indices.toArray, values.toArray)
}

// 转换目标字段，这里以Genres为例
val parsedDF = rawDF.withColumn("Genres", structToSparse(
  col("Genres.size"),
  col("Genres.indices"),
  col("Genres.values")
))

parsedDF.printSchema() // 此时Genres的类型会变成vector

场景2：Map是索引到值的直接映射（需要单独的维度字段） #

如果你的SparseVector是用{"0":1.0, "2":0.8}这种键值对存储，同时文档里有个字段（比如VectorSize）记录总维度，那UDF这么写：

import org.apache.spark.ml.linalg.SparseVector
import org.apache.spark.sql.functions.{udf, col}

val mapToSparse = udf { (vecMap: Map[String, Double], totalSize: Int) =>
  // 先把Map的键值对按索引排序，确保顺序正确
  val (indexStrs, values) = vecMap.toSeq.sortBy(_._1.toInt).unzip
  val indices = indexStrs.map(_.toInt).toArray
  new SparseVector(totalSize, indices, values.toArray)
}

// 转换DataFrame，假设总维度存在VectorSize字段
val parsedDF = rawDF.withColumn("Genres", mapToSparse(
  col("Genres"),
  col("VectorSize")
))

第三步：验证转换结果 #

转换完成后，你可以看看结果是否符合预期：

parsedDF.select("Title", "Genres").show(truncate = false)

这样处理后，原来的嵌套Map就变成了Spark MLlib能直接处理的Vector类型，不管是做后续的机器学习还是数据分析都没问题。如果有多个类似的SparseVector字段，复用这个UDF就行。

内容的提问来源于stack exchange，提问作者Daniil Andreyevich Baunov