PySpark如何读取CSV中带双重双引号的JSON字符串并提取字段

你的JSON解析结果返回null，是因为CSV存储时将JSON内部的双引号做了转义处理，变成了连续两个双引号，不符合标准JSON格式要求，json_tuple无法直接识别这种转义格式。

有两种常用的处理方案，你可以根据自己的场景选择：

• 方案1：加载CSV时配置转义规则，直接得到标准JSON字符串
  读取CSV时添加escape='"'参数，Spark会自动把转义的双引号还原为单个双引号，不需要额外处理字符串：

  from pyspark.sql import SparkSession
  import pyspark.sql.functions as F
  
  spark = SparkSession.builder.appName("parse_json_csv").getOrCreate()
  # 加载CSV时配置转义符
  df = spark.read.csv(
      "你的CSV文件路径",
      header=True,
      escape='"'
  )
  # 直接解析JSON即可
  result_df = df.select(
      F.json_tuple(F.col("request"), "CustomerId", "EffectiveTime")
      .alias("CustomerId", "EffectiveTime")
  )
  result_df.show()
  

• 方案2：已加载数据的情况下，先做字符串替换再解析
  如果已经完成数据加载，不想重新读取文件，可以先把request列的连续双引号替换为单个双引号，再做JSON解析：

  import pyspark.sql.functions as F
  from pyspark.sql.types import StructType, StructField, StringType
  
  # 第一步：清理转义字符
  df = df.withColumn("clean_request", F.regexp_replace(F.col("request"), '""', '"'))
  
  # 方法A：用json_tuple提取字段
  result_df = df.select(
      F.json_tuple(F.col("clean_request"), "CustomerId", "EffectiveTime")
      .alias("CustomerId", "EffectiveTime")
  )
  
  # 方法B：用from_json指定结构解析，更适合字段多、需要指定字段类型的场景
  json_schema = StructType([
      StructField("CustomerId", StringType(), True),
      StructField("EffectiveTime", StringType(), True)
  ])
  result_df = df.select(
      F.from_json(F.col("clean_request"), json_schema).alias("json_data")
  ).select("json_data.*")
  

两种方案最终输出的DataFrame都符合你需要的格式。

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