Spark用from_json解析JSON数组Schema时如何获取DataFrame的corrupt_record列

你直接使用ArrayType作为根Schema调用解析函数时，Spark没有预留字段存储解析失败的坏记录——corrupt_record字段必须定义在Schema的最外层Struct结构中，直接用ArrayType作为根层级没有位置存放该字段，自然无法捕获损坏记录。

采用两阶段解析逻辑，兼容数组格式的正常记录和非数组/结构不匹配的损坏记录：

• 第一阶段：初步解析JSON数组，识别整体格式非法的记录
  首先定义仅用于拆分数组的通用Schema，把数组内的每个元素先解析为字符串类型，避免直接解析业务字段导致整个数组解析失败：

  from pyspark.sql import functions as F
  from pyspark.sql.types import *
  
  # 数组拆分专用Schema：根为数组，数组元素为字符串类型
  array_split_schema = ArrayType(StringType())
  
  df = kafka_source_df.withColumn(
      "json_elements", 
      F.from_json(F.col("value"), array_split_schema)
  )
  

  这一步处理后：
  • 正常数组格式记录：json_elements列返回数组对象，每个元素是数组内的单条JSON字符串，比如示例正常记录会得到['{"event":"test","properties":{"p1":"v1"}}']
  • 非数组/非法JSON格式的损坏记录：json_elements列返回null，此时原始value列的内容就是损坏记录原文
• 第二阶段：解析数组内的业务字段，捕获单条元素级别的损坏记录
  定义带corrupt_record字段的业务Schema，注意该字段必须是StringType，且放在Schema最外层Struct中，符合Spark坏记录捕获规则：

  # 业务解析Schema，最外层显式加corrupt_record字段
  biz_schema = StructType([
      StructField("event", StringType(), True),
      StructField("properties", StructType([
          StructField("p1", StringType(), True)
      ]), True),
      StructField("corrupt_record", StringType(), True)
  ])
  

  先把合法数组炸开成单条记录做字段解析，再把第一阶段识别到的整体非法记录合并到结果中：

  # 处理合法数组内的单条元素
  parsed_valid_df = df.filter(F.col("json_elements").isNotNull()) \
      .select(F.explode(F.col("json_elements")).alias("single_json")) \
      .withColumn(
          "parsed",
          F.from_json(
              F.col("single_json"), 
              biz_schema,
              {"columnNameOfCorruptRecord": "corrupt_record"}
          )
      ) \
      .select("parsed.*")
  
  # 处理整体格式非法的损坏记录
  corrupt_array_df = df.filter(F.col("json_elements").isNull()) \
      .select(F.col("value").alias("corrupt_record")) \
      .withColumn("event", F.lit(None).cast(StringType())) \
      .withColumn("properties", F.lit(None).cast(biz_schema["properties"].dataType))
  
  # 合并得到带corrupt_record字段的完整结果
  final_df = parsed_valid_df.unionByName(corrupt_array_df)
  

用给出的示例数据校验解析结果：

• 正常记录[{"event":"test","properties":{"p1":"v1"}}]解析后，event值为test，properties.p1值为v1，corrupt_record为null
• 损坏记录{"corrupt_key":"value"}解析后，业务字段event、properties为null，corrupt_record字段值为原始损坏字符串
  如果不需要校验数组内单条元素的格式合法性，只需要判断整个Kafka消息是否符合数组结构要求，仅保留第一阶段逻辑，新增corrupt_record字段，当json_elements为null时取value列内容即可，无需执行第二阶段解析。

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