Spark Scala实现向量归一化及大数据集Z-score计算方法

Hey 👋！咱们逐个解决你的两个问题，都是针对Spark+Scala处理大数据场景的实用方案：

向量归一化常用的有三种，Spark MLlib都提供了开箱即用的分布式实现（强烈推荐用内置API，不要手动写循环/聚合，避免内存溢出）：

• L2归一化（向量单位化）：把向量缩放到模长为1，适合需要统一向量长度的场景（比如文本相似度计算）。
  代码示例：

  import org.apache.spark.sql.SparkSession
  import org.apache.spark.ml.feature.Normalizer
  
  val spark = SparkSession.builder().appName("VectorNormalization").getOrCreate()
  // 假设你的数据已经有一个名为"features"的向量列
  val df = spark.read.parquet("/path/to/your/data")
  
  // 初始化Normalizer，设置p=2就是L2归一化
  val normalizer = new Normalizer()
    .setInputCol("features")
    .setOutputCol("normalized_features")
    .setP(2.0)
  
  // 执行归一化
  val normalizedDF = normalizer.transform(df)
  

• Min-Max归一化（缩放到[0,1]区间）：把特征值映射到0到1之间，适合对数值范围敏感的模型。
  代码示例：

  import org.apache.spark.ml.feature.MinMaxScaler
  
  val scaler = new MinMaxScaler()
    .setInputCol("features")
    .setOutputCol("scaled_features")
    .setMin(0.0)
    .setMax(1.0)
  
  val scalerModel = scaler.fit(df)
  val scaledDF = scalerModel.transform(df)
  

• Z-Score标准化（均值为0，标准差为1）：就是你第二个问题里用到的公式，Spark用StandardScaler实现，后面第二个问题会详细讲。

先说说你现有代码的几个问题，这些问题在大数据场景下会直接导致性能崩溃或者内存溢出：

1. 你用了split(",")处理Parquet数据——完全没必要！Parquet是列式存储，直接读取列就行，不用像解析CSV那样拆分字符串。
2. result.sum / count这种方式会把所有数据的聚合结果拉到Driver节点，2000万条记录的话，Driver内存根本扛不住，而且是单节点计算，效率极低。
3. 代码里的sorted看起来是处理单维度数值，但你的数据集是id+features列，features如果是多维度向量的话，现有代码根本不适用。

下面给你两种适合大数据的方案：

方案一：Spark SQL分布式聚合（适合单特征场景） #

如果features是单个数值列（不是向量），用Spark SQL的内置聚合函数计算全局均值和标准差，全程分布式，不会把数据拉到Driver：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

val spark = SparkSession.builder().appName("ZScoreBigData").getOrCreate()
// 读取所有9个Parquet文件，Spark会自动加载路径下的所有Parquet文件
val df = spark.read.parquet("/path/to/your/parquet/files").select("id", "features") // 只选需要的列，减少数据传输

// 计算全局均值和标准差，用first()只把统计结果（两个数值）拉到Driver，不会拉全量数据
val stats = df.select(avg("features").alias("mean"), stddev("features").alias("stddev")).first()
val mean = stats.getDouble(0)
val stddev = stats.getDouble(1)

// 广播均值和标准差，让每个Executor都能复用，避免重复传输
val broadcastMean = spark.sparkContext.broadcast(mean)
val broadcastStddev = spark.sparkContext.broadcast(stddev)

// 计算Z-score，和你的代码保持一致用round取整
val zscoreDF = df.withColumn("zscore", round((col("features") - broadcastMean.value) / broadcastStddev.value))

// 保存结果到Parquet
zscoreDF.write.mode("overwrite").parquet("/path/to/save/zscore/result")

方案二：Spark MLlib StandardScaler（适合单/多特征向量场景） #

如果features是多维度向量列，或者想更优雅地处理（比如自动处理缺失值、分布式统计），用StandardScaler，它是专门为大数据设计的：

import org.apache.spark.sql.SparkSession
import org.apache.spark.ml.feature.{StandardScaler, VectorAssembler}

val spark = SparkSession.builder().appName("ZScoreMLlib").getOrCreate()
val df = spark.read.parquet("/path/to/your/parquet/files").select("id", "features")

// 如果features是单个数值列，先转成向量列（因为StandardScaler只处理向量）
val assembler = new VectorAssembler()
  .setInputCols(Array("features"))
  .setOutputCol("features_vec")

val vecDF = assembler.transform(df)

// 初始化StandardScaler，withMean=true才会执行(x-mean)/stddev，默认是只除以标准差
val scaler = new StandardScaler()
  .setInputCol("features_vec")
  .setOutputCol("scaled_features")
  .setWithMean(true) // 必须开这个才会减均值
  .setWithStd(true)

// 拟合数据，分布式计算全局均值和标准差，不会拉全量数据到Driver
val scalerModel = scaler.fit(vecDF)

// 转换数据得到Z-score向量
val scaledDF = scalerModel.transform(vecDF)

// 如果是单特征，把向量转成数值列，保留原列和id
val zscoreDF = scaledDF.withColumn("zscore", col("scaled_features").getItem(0))
  .select("id", "features", "zscore")

// 保存结果
zscoreDF.write.mode("overwrite").parquet("/path/to/save/zscore/result")

额外注意事项 #

• 缓存优化：如果需要多次使用中间结果，可以用df.cache()或df.persist(org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK)，避免重复读取Parquet。
• Parquet性能：读Parquet时可以开启spark.sql.parquet.enableVectorizedReader=true（默认开启），提升读取速度。
• 避免内存溢出：绝对不要用collect()把全量数据拉到Driver，只用first()拉统计结果。

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