时间戳

首页 > 解决方案 > 在pyspark中展平Maptype列

python-3.x - 在pyspark中展平Maptype列

问题描述

我有一个带有一列的 pyspark DataFrame，MapType并希望通过键名将其分解为所有列

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

我想做sp_test.select('a.*') 但得到一个错误：

AnalysisException: '只能用星号扩展结构数据类型。属性：ArrayBuffer(a);'

如果我们知道所有的键，这可以通过做

sp_test.select(['a.%s'%item for item in ['a','b']]).show()

但我想删除关键依赖

如果我们有一个 StrucType 列，这可以通过以下方式轻松实现display(nested_df.select('*', 'nested_array.*'))

root
 |-- _corrupt_record: string (nullable = true)
 |-- field1: long (nullable = true)
 |-- field2: long (nullable = true)
 |-- nested_array: struct (nullable = true)
 |    |-- nested_field1: long (nullable = true)
 |    |-- nested_field2: long (nullable = true)

我有些疑惑：

  1. 可以MapType转换为StructType?
  2. 我们可以直接从中查询子键MapType吗？

标签： python-3.xpyspark

解决方案

TL; DR：除非您提前知道可能的密钥，否则没有简单的方法可以完成您的要求。

让我用一个例子来解释为什么以及你的选择是什么。

首先，创建以下 DataFrame：

data = [({'a': 1, 'b': 2},), ({'c':3},), ({'a': 4, 'c': 5},)]
df = spark.createDataFrame(data, ["a"])
df.show()
#+-------------------+
#|                  a|
#+-------------------+
#|Map(a -> 1, b -> 2)|
#|        Map(c -> 3)|
#|Map(a -> 4, c -> 5)|
#+-------------------+

它具有以下架构：

df.printSchema()
#root
# |-- a: map (nullable = true)
# |    |-- key: string
# |    |-- value: long (valueContainsNull = true)
  1. 可以MapType转换为StructType?

简单的答案是否定的（至少效率不高），除非您提前知道密钥。

MapTypethe和 the之间的区别在于StructType映射的键值对是逐行独立的。StructType结构列中的列不是这种情况，所有行都具有相同的结构字段。

出于这个原因，spark 不能轻易地从地图中推断出要创建哪些列。（请记住，火花在每一行上并行运行）。另一方面，将结构分解成列很简单，因为所有列都是提前知道的。

因此，如果您知道密钥，则可以通过以下方式创建结构类型：

import pyspark.sql.functions as f

df_new = df.select(
    f.struct(*[f.col("a").getItem(c).alias(c) for c in ["a", "b", "c"]]).alias("a")
)
df_new.show()
#+-------------+
#|            a|
#+-------------+
#|   [1,2,null]|
#|[null,null,3]|
#|   [4,null,5]|
#+-------------+

新模式是：

df_new.printSchema()
#root
# |-- a: struct (nullable = false)
# |    |-- a: long (nullable = true)
# |    |-- b: long (nullable = true)
# |    |-- c: long (nullable = true)
  1. 我们可以直接从 MapType 查询子键吗？

是的，（如上所示）您可以使用getItem()which 将从列表中的索引处获取项目，或者通过键从地图中获取项目。

如果您不知道键，您唯一的选择是explode将映射成行，groupby并且pivot.

df.withColumn("id", f.monotonically_increasing_id())\
    .select("id", f.explode("a"))\
    .groupby("id")\
    .pivot("key")\
    .agg(f.first("value"))\
    .drop("id")\
    .show()
#+----+----+----+
#|   a|   b|   c|
#+----+----+----+
#|null|null|   3|
#|   1|   2|null|
#|   4|null|   5|
#+----+----+----+

在这种情况下，我们需要先创建一个id列，以便进行分组。

这里pivot可能很昂贵，具体取决于数据的大小。

推荐阅读