如何使用scala规范化或标准化spark中具有多列/变量的数据？

Question

我是 apache spark 和 scala 的新手。我有这样的数据集，我从 csv 文件中获取并使用 scala 将其转换为 RDD。

+-----------+-----------+----------+
| recent    | Freq      | Monitor  |
+-----------+-----------+----------+
|        1  |       1234 |   199090|
|        4  |       2553|    198613|
|        6  |       3232 |   199090|
|        1  |       8823 |   498831|
|        7  |       2902 |   890000|
|        8  |       7991 |   081097|
|        9  |       7391 |   432370|
|        12 |       6138 |   864981|
|        7  |       6812 |   749821|
+-----------+-----------+----------+

我想计算 z-score 值或标准化数据。所以我正在计算每列的 z 分数，然后尝试将它们组合起来，这样我就得到了标准比例。

这是我计算第一列 z 分数的代码

val scores1 = sorted.map(_.split(",")(0)).cache
val count = scores.count
val mean = scores.sum / count
val devs = scores.map(score => (score - mean) * (score - mean))
val stddev = Math.sqrt(devs.sum / count)
val zscore = sorted.map(x => math.round((x.toDouble - mean)/stddev)) 

如何计算每一列？或者有没有其他方法来规范化或标准化数据？

我的要求是分配等级（或等级）。

谢谢

Answer 1

如果要标准化列，可以使用 Spark MLlib 中的 StandardScaler 类。数据应采用RDD[Vectors[Double] 的形式，其中向量是MLlib Linalg 包的一部分。您可以选择使用均值或标准差或两者来标准化您的数据。

import org.apache.spark.mllib.feature.StandardScaler
import org.apache.spark.mllib.linalg.Vectors

val data = sc.parallelize(Array(
    Array(1.0,2.0,3.0),
    Array(4.0,5.0,6.0),
    Array(7.0,8.0,9.0),
    Array(10.0,11.0,12.0)))

// Converting RDD[Array] to RDD[Vectors]
val features = data.map(a => Vectors.dense(a))
// Creating a Scaler model that standardizes with both mean and SD
val scaler = new StandardScaler(withMean = true, withStd = true).fit(features)
// Scale features using the scaler model
val scaledFeatures = scaler.transform(features)

这个scaledFeatures RDD 包含所有列的 Z-score。

希望这个答案有帮助。查看文档以获取更多信息。

Answer 2

您可能希望使用以下代码对所需列执行标准缩放。向量汇编器用于选择需要转换的所需列。 StandardScaler 构造函数还为您提供了选择平均值和标准差值的选项

代码：

import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.feature
import org.apache.spark.ml.feature.StandardScaler

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val df = sqlContext.read.format("csv").option("header", "true").option("inferSchema", "true").load("/user/hadoop/data/your_dataset.csv")
df.show(Int.MaxValue)

val assembler = new VectorAssembler().setInputCols(Array("recent","Freq","Monitor")).setOutputCol("features")

val transformVector = assembler.transform(df)

val scaler = new StandardScaler().setInputCol("features").setOutputCol("scaledFeatures").setWithStd(true).setWithMean(false)

val scalerModel = scaler.fit(transformVector)
val scaledData = scalerModel.transform(transformVector)

scaledData.show() 20, False
scaledData.show(Int.MaxValue)
scaledData.show(20, false)