分解 Spark 列答案

【问题标题】：Factorize Spark column分解 Spark 列
【发布时间】：2016-09-28 18:18:23
【问题描述】：

是否可以分解 Spark 数据框列？分解我的意思是创建列中每个唯一值到相同 ID 的映射。

例如，原始数据框：

+----------+----------------+--------------------+
|      col1|            col2|                col3|
+----------+----------------+--------------------+
|1473490929|4060600988513370|                   A|
|1473492972|4060600988513370|                   A|
|1473509764|4060600988513370|                   B|
|1473513432|4060600988513370|                   C|
|1473513432|4060600988513370|                   A|
+----------+----------------+--------------------+

分解后的版本：

+----------+----------------+--------------------+
|      col1|            col2|                col3|
+----------+----------------+--------------------+
|1473490929|4060600988513370|                   0|
|1473492972|4060600988513370|                   0|
|1473509764|4060600988513370|                   1|
|1473513432|4060600988513370|                   2|
|1473513432|4060600988513370|                   0|
+----------+----------------+--------------------+

在 scala 本身中它会相当简单，但由于 Spark 将它的数据帧分布在节点上，我不确定如何保留来自 A->0, B->1, C->2 的映射。

另外，假设数据帧非常大（千兆字节），这意味着将一整列加载到单台机器的内存中可能是不可能的。

可以吗？

【问题讨论】：

标签： scala apache-spark spark-dataframe

【解决方案1】：

您可以使用StringIndexer 将字母编码为索引：

import org.apache.spark.ml.feature.StringIndexer

val indexer = new StringIndexer()
  .setInputCol("col3")
  .setOutputCol("col3Index")

val indexed = indexer.fit(df).transform(df)
indexed.show()

+----------+----------------+----+---------+
|      col1|            col2|col3|col3Index|
+----------+----------------+----+---------+
|1473490929|4060600988513370|   A|      0.0|
|1473492972|4060600988513370|   A|      0.0|
|1473509764|4060600988513370|   B|      1.0|
|1473513432|4060600988513370|   C|      2.0|
|1473513432|4060600988513370|   A|      0.0|
+----------+----------------+----+---------+

数据：

val df = spark.createDataFrame(Seq(
              (1473490929, "4060600988513370", "A"),
              (1473492972, "4060600988513370", "A"),  
              (1473509764, "4060600988513370", "B"),
              (1473513432, "4060600988513370", "C"),
              (1473513432, "4060600988513370", "A"))).toDF("col1", "col2", "col3")

【讨论】：

完美，就像一个魅力！完全忘记了 spark-mllib。

【解决方案2】：

您可以使用用户定义的函数。

首先创建所需的映射：

val updateFunction = udf {(x: String) =>
  x match {
    case "A" => 0
    case "B" => 1
    case "C" => 2
    case _ => 3
  }
}

现在您只需将其应用到您的DataFrame：

df.withColumn("col3", updateFunction(df.col("col3")))

【讨论】：

想象一下有 30 多个值 :)。维持这个将是一个地狱。
功能应该比这个好一点。这只是一个例子；）。
我知道，但是如果所有的值都是手动指定的，维护起来还是很困难的。