Scala Spark下适配MinHashLSH的DataFrame稀疏向量格式转换实现

之前自定义UDF写法的核心问题 #

• 链式调用逻辑失误：代码中val dfTransformed = df.withColumn(...).show() 直接将show()的返回值（Unit类型）赋值给了变量，根本无法拿到转换后的DataFrame实例，自然看不到预期结果。
• UDF入参类型不兼容：经过groupBy + collect_list聚合后，B列在Scala侧的实际类型是scala.collection.mutable.WrappedArray[Long]，而非代码中写的Array[Long]，硬编码入参类型会直接触发类型解析失败。
• 稀疏向量构造不合法：Spark MLlib构造SparseVector时强制要求传入的非零索引按从小到大升序排列，否则会抛出向量校验异常；而collect_list输出的元素顺序完全随机，没有天然排序逻辑，很容易构造出不符合规范的向量。
• 类型不匹配：如果UDF内错误使用了旧版mllib包下的Vector类型，会和MinHashLSH要求的ml.linalg.Vector类型不兼容，无法被MLlib组件正常识别。

可直接运行的实现代码 #

1. 提前导入需要的类：

import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.functions._
import org.apache.spark.ml.feature.MinHashLSH

2. 定义兼容类型、带合法性校验的UDF：

// 入参使用Seq[Long]，兼容collect_list输出的WrappedArray类型
val listToSparseVector = udf { valueList: Seq[Long] =>
  // 对集合内的元素去重、排序，保证索引严格升序，符合SparseVector构造规则
  val sortedIndex = valueList.map(_.toInt).distinct.sorted
  // 按目标格式构造9维稀疏向量，每个对应位置值为1.0
  Vectors.sparse(9, sortedIndex, Array.fill(sortedIndex.length)(1.0))
}

3. 执行列转换，拿到完整的转换后DataFrame：

val dfTransformed = df
  .withColumn("B", listToSparseVector(col("B")))

转换完成后可通过dfTransformed.show(false)校验输出结果，和手动构造的目标格式完全一致，可直接作为MinHashLSH组件的输入使用。

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