将 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。

但是，如果您提前知道键并且可以假设字符串始终采用这种格式，那么您应该能够使用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来测试字符串是否包含模式。

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