Spark from_json函数损坏记录处理及异常管控方案咨询

1. 给from_json添加错误处理并生成错误列 #

方法一：利用PERMISSIVE模式和columnNameOfCorruptRecord配置 #

from_json支持与JSON数据源一致的解析选项，要生成错误列，需同时配置mode="PERMISSIVE"和columnNameOfCorruptRecord，并将错误列字段加入解析schema中：

from pyspark.sql.functions import from_json, col
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("FromJsonErrorHandling").getOrCreate()

input_df = spark.sql("SELECT * FROM input_table")

# 扩展schema，加入错误列字段
json_schema = "struct<field1:string, field2:integer, corrupt_json:string>"
# 配置解析选项：宽松模式+指定错误列名
parse_options = {
    "mode": "PERMISSIVE",
    "columnNameOfCorruptRecord": "corrupt_json"
}

# 解析后得到包含正常字段和错误列的struct
output_df = input_df.withColumn("parsed_result", from_json(col("json_column"), json_schema, parse_options))

# 拆分struct为单独的解析列和错误列
final_df = output_df.select(
    "*",
    col("parsed_result.field1").alias("field1"),
    col("parsed_result.field2").alias("field2"),
    col("parsed_result.corrupt_json").alias("corrupt_json")
).drop("parsed_result")

final_df.show()

解析规则：

• 当JSON解析成功时，corrupt_json为null，field1/field2为正常解析值；
• 当JSON损坏时，field1/field2为null，corrupt_json存储原始损坏的JSON字符串；
• 当原始json_column为null时，所有字段均为null。

方法二：结合try()函数生成错误标记列 #

如果不需要保留原始损坏的JSON，仅需标记是否损坏，可以用try()包裹from_json，再通过条件判断生成错误标记列：

from pyspark.sql.functions import from_json, col, when, try_

input_df = spark.sql("SELECT * FROM input_table")

json_schema = "struct<field1:string, field2:integer>"

# try()会在解析失败时返回null
output_df = input_df.withColumn(
    "parsed_json",
    try_(from_json(col("json_column"), json_schema))
).withColumn(
    "is_corrupt",
    # 当解析结果为null且原始JSON不为null时，标记为损坏
    when((col("parsed_json").isNull()) & (col("json_column").isNotNull()), True).otherwise(False)
)

output_df.show()

2. 丢弃畸形记录 #

由于from_json不支持DROPMALFORMED模式，可基于上述错误处理的结果进行过滤：

基于方法一的过滤方式 #

过滤掉corrupt_json不为null的记录（即解析失败的畸形记录），保留原始JSON为null或解析成功的记录：

clean_df = final_df.filter(col("corrupt_json").isNull())

基于方法二的过滤方式 #

过滤掉is_corrupt为True的记录：

clean_df = output_df.filter(col("is_corrupt") == False)

两种方式都会保留原始json_column为null的记录，仅丢弃解析失败的畸形数据。

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