PySpark 将“地图”类型的列转换为数据框中的多列

Question

输入

我有一个列 Parameters 类型 map 的形式：

 from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
d = [{'Parameters': {'foo': '1', 'bar': '2', 'baz': 'aaa'}}]
df = sqlContext.createDataFrame(d)

df.collect()
# [Row(Parameters={'foo': '1', 'bar': '2', 'baz': 'aaa'})]

df.printSchema()
# root
#  |-- Parameters: map (nullable = true)
#  |    |-- key: string
#  |    |-- value: string (valueContainsNull = true)

输出

我想在 PySpark 中重塑它，以便所有键（ foo ， bar 等）都变成列，即：

 [Row(foo='1', bar='2', baz='aaa')]

使用 withColumn 有效：

 (df
 .withColumn('foo', df.Parameters['foo'])
 .withColumn('bar', df.Parameters['bar'])
 .withColumn('baz', df.Parameters['baz'])
 .drop('Parameters')
).collect()

但 我需要一个没有明确提及列名的解决方案，因为我有几十个。

原文由 Kamil Sindi 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

由于 MapType 的键不是模式的一部分，您必须首先收集这些键，例如：

 from pyspark.sql.functions import explode

keys = (df
    .select(explode("Parameters"))
    .select("key")
    .distinct()
    .rdd.flatMap(lambda x: x)
    .collect())

当你拥有这些时，剩下的就是简单的选择：

 from pyspark.sql.functions import col

exprs = [col("Parameters").getItem(k).alias(k) for k in keys]
df.select(*exprs)

原文由 zero323 发布，翻译遵循 CC BY-SA 3.0 许可协议

Answer 2

高性能解决方案

问题约束之一是动态确定列名，这很好，但请注意，这可能非常慢。以下是如何避免键入并编写可快速执行的代码。

 cols = list(map(
    lambda f: F.col("Parameters").getItem(f).alias(str(f)),
    ["foo", "bar", "baz"]))
df.select(cols).show()

 +---+---+---+
|foo|bar|baz|
+---+---+---+
|  1|  2|aaa|
+---+---+---+

请注意，这将运行单个选择操作。不要运行 withColumn 多次，因为这样会更慢。

快速解决方案只有在您知道所有地图键的情况下才有可能。如果您不知道映射键的所有唯一值，则需要恢复到较慢的解决方案。

较慢的解决方案

接受的答案很好。我的解决方案性能更高一些，因为它不调用 .rdd 或 flatMap() 。

 import pyspark.sql.functions as F

d = [{'Parameters': {'foo': '1', 'bar': '2', 'baz': 'aaa'}}]
df = spark.createDataFrame(d)

keys_df = df.select(F.explode(F.map_keys(F.col("Parameters")))).distinct()
keys = list(map(lambda row: row[0], keys_df.collect()))
key_cols = list(map(lambda f: F.col("Parameters").getItem(f).alias(str(f)), keys))
df.select(key_cols).show()

 +---+---+---+
|bar|foo|baz|
+---+---+---+
|  2|  1|aaa|
+---+---+---+

将结果收集到驱动程序节点可能是性能瓶颈。最好将此代码 list(map(lambda row: row[0], keys_df.collect())) 作为一个单独的命令执行，以确保它不会运行得太慢。

原文由 Powers 发布，翻译遵循 CC BY-SA 4.0 许可协议