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

Question

我有一个数据框（在 Pyspark 中），其中一个行值作为字典：

df.show()

它看起来像：

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

基于cmets给予更多：

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

正如您使用 printSchema 函数所见，“Spark”将您的字典理解为字符串。分割字符串并创建新列的函数是 split() ，因此可以简单地解决这个问题。

• 创建一个能够：

  • 将字典字符串转换为逗号分隔的字符串（从字典中删除键，但保持值的顺序）

• 应用拆分并根据我们字典的新格式创建两个新列

代码：

@udf()
def transform_dict(dict_str):
    str_of_dict_values = dict_str.\
        replace("}", "").\
        replace("{", ""). \
        replace("color:", ""). \
        replace(" car: ", ""). \
        strip()
    # output example: 'red,volkswagen'
    return str_of_dict_values

# Create new column with our UDF with the dict values converted to str
df = df.withColumn('info_clean', clean("info"))
# Split these values and store in a tmp variable 
split_col = split(df['info_clean'], ',')

# Create new columns with the split values
df = df.withColumn('color', split_col.getItem(0))
df = df.withColumn('car', split_col.getItem(1))

只有当我们假设字典元素总是以相同的顺序出现并且键是固定的时，这个解决方案才是正确的。 对于其他更复杂的情况，我们可以在 UDF 函数中创建一个字典，并通过显式调用每个字典键来形成值列表字符串，因此我们将确保输出链中的顺序保持不变。

Answer 2

你的字符串：

"{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 来测试字符串是否包含该模式，然后再尝试提取匹配项。

---

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