Spark：将 RDD[LabeledPoint] 转换为 Dataframe 以应用 MinMaxScaler，缩放后得到归一化的 RDD[LabeledPoint]答案

【问题标题】：Spark: convert an RDD[LabeledPoint] to a Dataframe to apply MinMaxScaler, and after scaling get the normalized RDD[LabeledPoint]Spark：将 RDD[LabeledPoint] 转换为 Dataframe 以应用 MinMaxScaler，缩放后得到归一化的 RDD[LabeledPoint]
【发布时间】：2018-10-20 19:03:05
【问题描述】：

我在我的代码中使用 RDD[LabeledPoint]。但现在我必须使用 MinMax 方法对数据进行标准化。

我看到 ml 库中存在 MinMaxScaler，但这适用于 DataFrames：org.apache.spark.ml.feature.MinMaxScaler。

由于已经用 RDD 编写了完整的代码，我想我可以执行以下步骤来不更改任何其他内容：

将 RDD[LabeledPoint] 转换为 DataFrame
将 MinMaxScaler 应用于 DataFrame
将 DataFrame 转换为 RDD[LabeledPoint]

问题是我不知道我怎么能做到。我没有列名（但 LabeledPoint 中的特征向量有 9 维），我也无法根据我的情况调整其他示例。例如，中的代码： https://stackoverflow.com/a/36909553/5081366 或Scaling each column of a dataframe

感谢您的帮助！

【问题讨论】：

标签： scala apache-spark

【解决方案1】：

终于可以回答我自己的问题了！

其中allData 是RDD[LabeledPoint]：

    // The following import doesn't work externally because the implicits object is defined inside the SQLContext class
    val sqlContext = SparkSession
      .builder()
      .appName("Spark In Action")
      .master("local")
      .getOrCreate()

    import sqlContext.implicits._

    // Create a DataFrame from RDD[LabeledPoint]
    val all = allData.map(e => (e.label, e.features))
    val df_all = all.toDF("labels", "features")

    // Scaler instance above with the same min(0) and max(1)
    val scaler = new MinMaxScaler()
      .setInputCol("features")
      .setOutputCol("featuresScaled")
      .setMax(1)
      .setMin(0)

    // Scaling
    var df_scaled = scaler.fit(df_all).transform(df_all)

    // Drop the unscaled column
    df_scaled = df_scaled.drop("features")

    // Convert DataFrame to RDD[LabeledPoint]
    val rdd_scaled = df_scaled.rdd.map(row => LabeledPoint(
      row.getAs[Double]("labels"),
      row.getAs[Vector]("featuresScaled")
    ))

我希望这对其他人有帮助！

【讨论】：