如何在PySpark DataFrame中转换列内JSON字符串并提取键值对

没问题，完全不用转Pandas，PySpark本身就有高效的内置工具处理这种JSON字符串列，而且性能比用UDF或者转Pandas好太多，刚好适合大数据场景。咱们一步步来：

1. 先定义JSON对应的Schema #

因为你的_c0列是结构化的JSON字符串，PySpark需要明确的Schema来正确解析它（避免自动推断带来的性能损耗）。根据你给出的示例，我们可以定义如下Schema：

from pyspark.sql.types import StructType, StructField, StringType, ArrayType, FloatType

# 匹配JSON结构的Schema
json_schema = StructType([
    StructField("object", StringType(), nullable=True),
    StructField("time", StringType(), nullable=True),  # 如果需要直接转时间戳，换成TimestampType()即可
    StructField("values", ArrayType(FloatType()), nullable=True)
])

2. 解析JSON字符串列 #

用PySpark内置的from_json函数把_c0的字符串转成结构化的列，之后就可以轻松提取各个键值对了：

from pyspark.sql.functions import from_json, col

# 解析_c0列，生成一个包含结构化数据的新列
parsed_df = df.withColumn("parsed_data", from_json(col("_c0"), json_schema))

# 把解析后的字段提取成单独的列，方便后续使用
result_df = parsed_df.select(
    col("_c1"),  # 保留原来的_c1列
    col("parsed_data.object").alias("object"),
    col("parsed_data.time").alias("time"),
    col("parsed_data.values").alias("values")
)

# 查看结果
result_df.show(truncate=False)

3. 提取字段到变量 #

如果需要把这些字段的值存储到变量里，比如提取某一列的所有值，或者某一行的具体数据，可以这样做：

# 提取所有object字段的值到列表变量
all_objects = [row["object"] for row in result_df.select("object").collect()]

# 提取第一行的values数组到变量
first_row_values = result_df.select("values").first()[0]

# 如果要统计values的平均值，直接用PySpark的聚合函数，不用转成Python变量
from pyspark.sql.functions import array_avg
avg_value = result_df.select(array_avg(col("values"))).first()[0]

小提示 #

• 尽量用PySpark的内置函数，别写自定义UDF，内置函数是经过优化的，在大数据量下性能差很多
• 如果你的JSON结构有变动，记得同步调整Schema，不然会解析失败或者丢失字段

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