Scala Spark代码报错Unsupported literal type DenseVector 求助

错误原因 #

Spark的lit()函数仅支持基本数据类型（如Int、String、Double），不直接支持org.apache.spark.ml.linalg.DenseVector这类MLlib复杂类型。即使移除lit()，直接将本地的DenseVector对象传入when分支，Spark仍会尝试将其解析为字面量，同样触发不支持类型的错误。

解决方案 #

方法1：使用typedLit()（Spark 2.2+推荐） #

typedLit()是Spark专门为复杂类型（向量、数组、自定义类等）设计的字面量构造函数，直接替换原代码中的lit()即可：

import org.apache.spark.sql.functions.typedLit
import org.apache.spark.ml.linalg.DenseVector

val joinedDf = sampleDf
  .join(
    normalizedParentEmbedding.select(col("parent_product"), col("normalized_vector").as("parent_vector")),
    Seq("parent_product"),
    "left"
  )
  .join(
    normalizedChildEmbeddingDf.select(col("child_product"), col("normalized_vector").alias("child_vector")),
    Seq("child_product"),
    "left"
  )
  .withColumn("parent_vector", when(col("parent_vector").isNull,
         typedLit(normalizedDefaultParentEmbeddingDf.select("parent_vector").head()(0).asInstanceOf[DenseVector]))
         .otherwise(col("parent_vector")))
  .withColumn("child_vector", when(col("child_vector").isNull,                
         typedLit(normalizedDefaultChildEmbeddingDf.select("child_vector").head()(0).asInstanceOf[DenseVector]))
         .otherwise(col("child_vector")))

方法2：通过广播单行DataFrame引入默认值（兼容低版本Spark） #

如果你的Spark版本低于2.2，可先将默认向量封装为单行DataFrame，通过广播关联后用coalesce替换空值：

import org.apache.spark.ml.linalg.DenseVector
import org.apache.spark.sql.functions.{broadcast, coalesce}

// 获取默认向量并创建单行DataFrame
val defaultParentVec = normalizedDefaultParentEmbeddingDf.select("parent_vector").head()(0).asInstanceOf[DenseVector]
val defaultParentDf = spark.createDataFrame(Seq(defaultParentVec)).toDF("default_parent_vec")

val defaultChildVec = normalizedDefaultChildEmbeddingDf.select("child_vector").head()(0).asInstanceOf[DenseVector]
val defaultChildDf = spark.createDataFrame(Seq(defaultChildVec)).toDF("default_child_vec")

// 执行关联与空值替换
val joinedDf = sampleDf
  .join(
    normalizedParentEmbedding.select(col("parent_product"), col("normalized_vector").as("parent_vector")),
    Seq("parent_product"),
    "left"
  )
  .join(
    normalizedChildEmbeddingDf.select(col("child_product"), col("normalized_vector").alias("child_vector")),
    Seq("child_product"),
    "left"
  )
  .crossJoin(broadcast(defaultParentDf))
  .crossJoin(broadcast(defaultChildDf))
  .withColumn("parent_vector", coalesce(col("parent_vector"), col("default_parent_vec")))
  .withColumn("child_vector", coalesce(col("child_vector"), col("default_child_vec")))
  .drop("default_parent_vec", "default_child_vec")

关键说明 #

• typedLit()会保留复杂类型的结构信息，让Spark正确识别DenseVector类型
• 广播单行DataFrame避免全量数据的笛卡尔积开销，保证性能

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