Scala中DataFrame用Embedding填充Null时出现数据类型不匹配错误

问题 #

我有父产品64维嵌入向量和子产品64维嵌入向量的DataFrame，每行包含产品ID与64个嵌入元素，另有同结构的默认嵌入向量DataFrame。

已通过以下代码完成嵌入向量的组装与归一化：

def normalizeEmbeddings(df: DataFrame, embeddingSize: Int = 64): DataFrame = {
    val denseVectorDf = new VectorAssembler()
      .setInputCols((0.until(embeddingSize)).toArray.map(_.toString))
      .setOutputCol("vector")
      .transform(df)
    new Normalizer().setInputCol("vector").setOutputCol("normalized_vector").setP(2).transform(denseVectorDf)
  }

val normalizedParentEmbeddingDf = normalizeEmbeddings(parentEmbeddingDf)
val normalizedChildEmbeddingDf = normalizeEmbeddings(childEmbeddingDf)
val normalizedDefaultParentEmbeddingDf = normalizeEmbeddings(defaultParentEmbedding)  // one row
val normalizedDefaultChildEmbeddingDf = normalizeEmbeddings(defaultChildEmbedding)  // one row

随后将嵌入向量DataFrame与样本数据关联，逻辑为无嵌入向量的产品使用默认值，但执行以下填充代码时出现错误：

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, normalizedDefaultParentEmbeddingDf.select("parent_vector").head()(0)).otherwise(col("parent_vector")))  // ERROR!
  .withColumn("child_vector", when(col("child_vector").isNull, normalizedDefaultChildEmbeddingDf.select("child_vector").head()(0)).otherwise(col("child_vector")))

错误信息如下：

org.apache.spark.sql.AnalysisException: cannot resolve 'CASE WHEN (`parent_vector` IS NULL) THEN 'default' ELSE `parent_vector` END' due to data type mismatch: THEN and ELSE expressions should all be same type or coercible to a common type;;

请问该如何解决此数据类型不匹配问题？

解决方法 #

问题核心是直接调用head()(0)获取默认向量时，会把本地Scala对象（如DenseVector）传入Spark Column表达式，Spark无法识别其类型，导致与目标列的Vector类型不匹配。以下是两种可行的修正方案：

方案1：用lit()包装默认向量为常量列 #

先提取默认向量的本地值，再用lit()转换为Spark可识别的Column常量，确保类型统一：

// 提取默认向量并转为Spark Column兼容的常量
val defaultParentVec = normalizedDefaultParentEmbeddingDf.select("normalized_vector").head().getAs[org.apache.spark.ml.linalg.Vector](0)
val defaultChildVec = normalizedDefaultChildEmbeddingDf.select("normalized_vector").head().getAs[org.apache.spark.ml.linalg.Vector](0)

val joinedDf = sampleDf
  .join(
    normalizedParentEmbeddingDf.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, lit(defaultParentVec)).otherwise(col("parent_vector")))
  .withColumn("child_vector", when(col("child_vector").isNull, lit(defaultChildVec)).otherwise(col("child_vector")))

方案2：用crossJoin关联默认向量+coalesce填充 #

避免直接读取本地数据，通过笛卡尔积关联默认向量DataFrame，再用coalesce()取第一个非空值，逻辑更简洁：

val joinedDf = sampleDf
  .join(
    normalizedParentEmbeddingDf.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(normalizedDefaultParentEmbeddingDf.select(col("normalized_vector").as("default_parent_vec")))
  .crossJoin(normalizedDefaultChildEmbeddingDf.select(col("normalized_vector").as("default_child_vec")))
  .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")

关键注意点 #

• lit()是Spark将本地对象转为Column常量的标准方法，能保证类型与目标列一致
• coalesce()可直接返回第一个非空列，适合多来源的默认值填充场景
• 禁止在Column表达式中直接调用head()这类触发Action的方法，容易引发类型不匹配或分布式执行逻辑错误

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