Spark Scala如何在map函数内使用if语句实现数值转换

你的代码跑不通是三个低级错误导致的：

• row.getAs[DenseVector]("vScaled").values 返回的是Array[Double]类型，数组对象本身不能直接和Double类型的0、1做大小比较，数组 > 1这种写法本身就不符合Scala语法规则。
• 判断逻辑写错了维度：你要处理的是数组里的每一个元素，不是拿整个数组和阈值比、再把整个数组替换成单个值的数组。
• 直接修改Row中取出的DenseVector对象内部的values数组属于原地修改，在Spark分布式计算场景下很容易触发不可变对象异常、序列化错误，本身就不推荐这么写。

方案1：直接修正RDD map逻辑，完全匹配你原有代码的流程 #

import org.apache.spark.ml.linalg.DenseVector
import org.nd4j.linalg.factory.Nd4j

val df2 = df1.select("vScaled")
val sqldf = df2.rdd
  .map { row =>
    val rawValues = row.getAs[DenseVector]("vScaled").values
    // 逐元素做[0,1]区间截断
    rawValues.map { element =>
      if (element > 1.0) 1.0
      else if (element < 0.0) 0.0
      else element
    }
  }
  .map(processedArr => Nd4j.createFromArray(Array(processedArr)))

方案2：用DataFrame原生UDF先做截断，性能更好 #

DataFrame的原生算子有Catalyst优化，比手写RDD map执行效率更高，适合数据量较大的场景：

import org.apache.spark.ml.linalg.DenseVector
import org.apache.spark.sql.functions.{col, udf}
import org.nd4j.linalg.factory.Nd4j

// 定义向量截断UDF，逐元素把值限制在0-1区间
val clipVecUdf = udf((vec: DenseVector) => {
  new DenseVector(vec.values.map(e => math.min(1.0, math.max(0.0, e))))
})

val sqldf = df1.select(clipVecUdf(col("vScaled")).alias("vScaled"))
  .rdd
  .map(row => row.getAs[DenseVector]("vScaled").values)
  .map(arr => Nd4j.createFromArray(Array(arr)))

注：代码里math.min(1.0, math.max(0.0, e))和if-else判断逻辑完全等价，是数值截断的常用简洁写法。

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