python - 将 Pyspark 数据框中的字典拆分为单独的列

Question

我有一个数据框(在 Pyspark 中)，其中一个行值作为字典:

df.show()

看起来像:

+----+---+-----------------------------+
|name|age|info                         |
+----+---+-----------------------------+
|rob |26 |{color: red, car: volkswagen}|
|evan|25 |{color: blue, car: mazda}    |
+----+---+-----------------------------+

根据评论给出更多:

df.printSchema()

类型是字符串

root
 |-- name: string (nullable = true)
 |-- age: string (nullable = true)
 |-- dict: string (nullable = true)

是否可以从字典中获取键(颜色和汽车)并将它们设为数据框中的列，并将值作为这些列的行？

预期结果:

+----+---+-----------------------------+
|name|age|color |car                   |
+----+---+-----------------------------+
|rob |26 |red   |volkswagen            |
|evan|25 |blue  |mazda                 |
+----+---+-----------------------------+

我不知道我必须使用 df.withColumn() 并以某种方式遍历字典来选择每个字典，然后从中创建一列？到目前为止，我已经尝试找到一些答案，但大多数都使用 Pandas，而不是 Spark，所以我不确定是否可以应用相同的逻辑。

Answer 1

你的字符串:

"{color: red, car: volkswagen}"
"{color: blue, car: mazda}"

不是Python友好的格式。它们无法使用 json.loads 进行解析，也不能使用 ast.literal_eval 进行评估。

但是，如果您提前知道 key 并且可以假设字符串始终采用这种格式，那么您应该能够使用 pyspark.sql.functions.regexp_extract :

例如:

from pyspark.sql.functions import regexp_extract

df.withColumn("color", regexp_extract("info", "(?<=color: )\w+(?=(,|}))", 0))\
    .withColumn("car", regexp_extract("info", "(?<=car: )\w+(?=(,|}))", 0))\
    .show(truncate=False)
#+----+---+-----------------------------+-----+----------+
#|name|age|info                         |color|car       |
#+----+---+-----------------------------+-----+----------+
#|rob |26 |{color: red, car: volkswagen}|red  |volkswagen|
#|evan|25 |{color: blue, car: mazda}    |blue |mazda     |
#+----+---+-----------------------------+-----+----------+

模式是:

• (?<=color: ) :文字字符串 "color: " 的正向后查找
• \w+ :一个或多个单词字符
• (?=(,|})) :对原义逗号或右大括号进行积极的前瞻。

<小时/>

以下是如何将其推广到两个以上的键，并处理字符串中不存在该键的情况。

from pyspark.sql.functions import regexp_extract, when, col
from functools import reduce

keys = ["color", "car", "year"]
pat = "(?<=%s: )\w+(?=(,|}))"

df = reduce(
    lambda df, c: df.withColumn(
        c,
        when(
            col("info").rlike(pat%c),
            regexp_extract("info", pat%c, 0)
        )
    ),
    keys,
    df
)

df.drop("info").show(truncate=False)
#+----+---+-----+----------+----+
#|name|age|color|car       |year|
#+----+---+-----+----------+----+
#|rob |26 |red  |volkswagen|null|
#|evan|25 |blue |mazda     |null|
#+----+---+-----+----------+----+

在本例中，我们使用 pyspark.sql.functions.when和pyspark.sql.Column.rlike在我们尝试提取匹配之前测试字符串是否包含模式。

<小时/>

如果您事先不知道 key ，则必须编写自己的解析器或尝试修改上游数据。