带有虚拟变量的 pyspark 矩阵答案

【问题标题】：pyspark matrix with dummy variables带有虚拟变量的 pyspark 矩阵
【发布时间】：2016-03-08 22:33:20
【问题描述】：

有两列：

ID  Text
1    a
2    b
3    c

我怎样才能用这样的虚拟变量创建矩阵：

使用 pyspark 库及其功能？

【问题讨论】：

标签： python apache-spark pyspark

【解决方案1】：

另一种解决方案是使用 Spark 的 pivot 方法，该方法自 Spark 1.6.0 以来一直存在。

例子：

from pyspark.sql import functions as F

df = sqlContext.createDataFrame([
    (1, "a"),
    (2, "b"),
    (3, "c"),],
    ["ID", "Text"])    

pivoted = df.groupBy("ID").pivot("Text").agg(F.lit(1))
pivoted.show()
# +---+----+----+----+
# | ID|   a|   b|   c|
# +---+----+----+----+
# |  1|   1|null|null|
# |  3|null|null|   1|
# |  2|null|   1|null|
# +---+----+----+----+

要消除缺失值，只需使用na 方法：

pivoted.na.fill(0).show()
# +---+---+---+---+
# | ID|  a|  b|  c|
# +---+---+---+---+
# |  1|  1|  0|  0|
# |  3|  0|  0|  1|
# |  2|  0|  1|  0|
# +---+---+---+---+

Pivoting 比 solution proposed by ksindi 更通用，因为它可以聚合数字。话虽这么说，ksindi 提出的解决方案在这种特殊情况下更有效，因为它只需要对数据进行一次传递，如果考虑到获取类别的传递，则需要两次。对于旋转，您还可以将类别作为第二个位置参数添加到pivot，从而提高效率。但是groupBy 调用已经引起了洗牌，这使得这种方法变慢了。

注意：groupBy 调用默认假定示例中的ID 列包含唯一值以获取所需的输出。如果示例数据框看起来像：

df = sqlContext.createDataFrame([
    (1, "a"),
    (2, "b"),
    (3, "c"),
    (3, "a")],
    ["ID", "Text"])

这个解决方案的结果是

df.groupBy("ID").pivot("Text").agg(F.lit(1)).na.fill(0).show()
# +---+---+---+---+
# | ID|  a|  b|  c|
# +---+---+---+---+
# |  1|  1|  0|  0|
# |  3|  1|  0|  1|
# |  2|  0|  1|  0|
# +---+---+---+---+

而映射解决方案最终会是

df.select("ID", *exprs).show()
# +---+---+---+---+
# | ID|  c|  b|  a|
# +---+---+---+---+
# |  1|  0|  0|  1|
# |  2|  0|  1|  0|
# |  3|  1|  0|  0|
# |  3|  0|  0|  1|
# +---+---+---+---+

【讨论】：

这种方法对我来说更干净，但也想在制作虚拟列时报告原始列名，例如列“a”、“b”、“c”应该变成@ 987654334@。例如，如果您需要从多个列创建虚拟编码，您可能会无意中替换旧列。另外，您应该指出 F.monotonically_increasing_id 将有助于区分行，充当 .groupBy 表达式。

【解决方案2】：

from pyspark.sql import functions as F

df = sqlContext.createDataFrame([
    (1, "a"),
    (2, "b"),
    (3, "c"),
], ["ID", "Text"])

categories = df.select("Text").distinct().rdd.flatMap(lambda x: x).collect()

exprs = [F.when(F.col("Text") == category, 1).otherwise(0).alias(category)
         for category in categories]

df.select("ID", *exprs).show()

输出

+---+---+---+---+
| ID|  a|  b|  c|
+---+---+---+---+
|  1|  1|  0|  0|
|  2|  0|  1|  0|
|  3|  0|  0|  1|
+---+---+---+---+

【讨论】：

回溯（最近一次调用最后一次）：文件“”，第 2 行，在 NameError: name 'F' is not defined
这里的 F 是什么？
从 pyspark.sql 导入函数为 F
谢谢你——如何将最终的选择语句“df.select("ID", *exprs).show()" 与列列表而不是单个字符串结合起来引用列？例如，如果我想做 "df.select(df.columns, *exprs).show()" 我会收到错误消息： Invalid argument, not a string or column: ['column names...'] 类型。对于列文字，请使用“lit”、“array”、“struct”或“create_map”函数……有人能指出我正确的方向吗？谢谢。

【解决方案3】：

如果虚拟变量用于建模，您可以使用 OneHotEncoder /OneHotEncoderEstimator（取决于 pyspark 的版本）。

可以在Spark MLlib Guide中展示一个示例。

请注意，如果您需要删除一列以防止线性依赖（如回归的情况），您应该使用dropLast=True。

【讨论】：