将字符串列转换为JSON并在PySpark中解析

在Py Spark 中，我们可以使用自定义 函数 （UDF）将字符串列转换为JSON，并使用内置 函数 将JSON 解析 为所需的结构。

首先，我们需要使用内置 函数 from_json 将JSON字符串 解析 为结构化数据。

from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType
# 定义JSON格式
json_schema = StructType([
  StructField("name", StringType()),
  StructField("age", StringType()),
  StructField("gender", StringType())
# 解析JSON字符串为结构化数据
df = df.withColumn("json_data", from_json(col("json_str"), json_schema))
接下来，我们可以使用内置函数getItem访问JSON对象中的特定属性。
# 获取JSON中的特定属性
df = df.withColumn("name", col("json_data").getItem("name"))
df = df.withColumn("age", col("json_data").getItem("age"))
df = df.withColumn("gender", col("json_data").getItem("gender"))
# 删除JSON列
df = df.drop("json_data")
最后，我们可以将UDF应用于字符串列，并将结果转换为所需的数据类型。
import json
# 定义UDF并将其应用于字符串列
@udf("string")
def parse_json(json_str):
    json_obj = json.loads(json_str)
    return json_obj["id"]
# 将结果转换为所需的数据类型
df = df.withColumn("id", parse_json(col("json_str")).cast("integer"))
# 删除原始字符串列
df = df.drop("json_str")