PySpark如何将CSV中字符串格式的映射列转换为map类型

1. 读取CSV文件 #

读取时配置引号解析规则，正确识别被双引号包裹的整列值：

from pyspark.sql import SparkSession
from pyspark.sql.functions import regexp_replace, from_json, concat, lit
from pyspark.sql.types import MapType, StringType

# 初始化SparkSession
spark = SparkSession.builder.appName("csv_to_map").getOrCreate()

# 读取CSV文件，quote参数指定双引号为值包裹符，会自动去掉值外层多余的双引号
df = spark.read.csv(
    "你的CSV文件路径.csv",
    header=True,
    quote='"',
    inferSchema=False
)

读取完成后name_value列的原始值格式为 [quality1 -> good, quality2 -> OK, quality3 -> bad]。

2. 转换为Map类型 #

核心思路是把自定义格式的字符串转换为标准JSON对象格式，再用from_json函数直接解析为Map类型：

df_with_map = df.withColumn(
    "name_value_map",
    from_json(
        # 拼接生成标准JSON字符串
        concat(
            lit('{"'),
            # 三层替换逻辑：先去掉首尾方括号，再把->替换为JSON键值分隔符，最后把逗号分隔符转为JSON键值对分隔符
            regexp_replace(
                regexp_replace(
                    regexp_replace("name_value", r"^\[|\]$", ""),
                    "->", '":"'
                ),
                ", ", '", "'
            ),
            lit('"}')
        ),
        # 指定解析后Map的键和值都为字符串类型
        MapType(StringType(), StringType())
    )
)

结果验证 #

可以直接调用map的键取值，确认转换是否成功：

# 查看完整map列
df_with_map.select("name_value_map").show(truncate=False)

# 访问map指定键对应的值
df_with_map.select(df_with_map.name_value_map["quality1"]).show()

内容的提问来源于stack exchange，提问作者Arthur