如何指定PySpark DataFrame列类型为JSON实现无转义嵌套输出

实现思路 #

由于单批次运行时requestBody的JSON结构固定，我们可以先从解密后的JSON字符串样本自动推断schema，再将字符串类型的requestBody转为Spark结构化类型，写入时就会自动输出为无转义的嵌套JSON对象。

具体操作步骤 #

• 第一步：推断requestBody的动态schema
  两种可选方案，根据场景选择：
  方案1：基于小样本推断（兼容性最高，适合字段可能有缺省的场景）

  # 取少量样本即可覆盖单批次固定schema，样本量可按需调整
  sample_rdd = df_decrypted.select("requestBody").limit(100).rdd.map(lambda x: x.requestBody)
  request_body_schema = spark.read.json(sample_rdd).schema
  

  方案2：基于单条有效数据推断（更轻量，适合所有数据结构完全一致的场景）

  from pyspark.sql.functions import col
  
  first_valid_body = df_decrypted.filter(col("requestBody").isNotNull()).first().requestBody
  request_body_schema = spark.sql(f"SELECT schema_of_json('{first_valid_body}')").head()[0]
  

• 第二步：将JSON字符串转为结构化类型
  使用from_json函数结合推断出的schema，把字符串类型的requestBody转成Spark原生结构化类型：

  from pyspark.sql.functions import from_json
  
  df_final = df_decrypted.withColumn("requestBody", from_json("requestBody", request_body_schema))
  

• 第三步：写入S3
  按原有逻辑写入JSON格式即可，此时requestBody会作为嵌套JSON对象输出，不存在转义符：

  df_final.write.mode("overwrite").json("s3://your-bucket/output-path/")
  

异常处理建议 #

如果存在解密后requestBody格式非法的场景，可以在from_json调用时添加参数控制解析行为：

from pyspark.sql.functions import from_json, lit

df_final = df_decrypted.withColumn(
    "requestBody",
    from_json("requestBody", request_body_schema, options={"mode": "PERMISSIVE"})
)
# 非法JSON解析后会返回null，可按需过滤或保留

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