PySpark将"map"类型的列转换为数据框中的多个列

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'})]

产量

我想重塑它在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()

但我需要一个没有明确提到列名的解决方案,因为我有几十个.

架构

>>> df.printSchema()

root
 |-- Parameters: map (nullable = true)
 |    |-- key: string
 |    |-- value: string (valueContainsNull = true)

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)

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多次运行，因为那样会比较慢。

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

较慢的解决方案

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

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

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

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