如何高效处理拼接错误的JSON文件？

一、PySpark DataFrame直接处理的方案 #

Spark默认的JSON数据源仅支持合法的JSON数组，或每行一个独立JSON对象的JSON Lines格式，无法直接解析这种多根对象拼接的错误格式。但可以通过分布式文本处理来解决：

分布式拆分方案（适合超大文件） #

利用Spark RDD的分布式特性，避免将全量数据拉到Driver节点，步骤如下：

def split_json_partition(iterator):
    # 处理每个分区的文本块
    for part in iterator:
        # 用"}\n{"替换"}{"来拆分多个JSON对象
        split_items = part.replace("}{", "}\n{").split("\n")
        for json_str in split_items:
            if json_str.strip():  # 过滤空行和空白内容
                yield json_str

# 读取文本文件，按文件大小设置合理分区数
text_rdd = spark.sparkContext.textFile("path/to/your/invalid.json", minPartitions=8)
# 将每个分区内的所有行拼接成完整文本块
partitioned_text = text_rdd.mapPartitions(lambda iter: ["".join(iter)])
# 拆分每个分区的JSON对象并扁平化
json_rdd = partitioned_text.flatMap(split_json_partition)
# 转为DataFrame
df = spark.read.json(json_rdd)

二、Python预处理修复（适合中小文件） #

如果文件体积不大，直接用Python修改格式为合法JSON数组是最简便的方式：

# 读取错误格式的JSON文件
with open("invalid.json", "r", encoding="utf-8") as f:
    raw_content = f.read()

# 修复格式：包裹为JSON数组，替换对象分隔符
fixed_content = "[" + raw_content.replace("}{", "}, {") + "]"

# 保存修复后的文件
with open("fixed.json", "w", encoding="utf-8") as f:
    f.write(fixed_content)

# 正常读取修复后的文件
df = spark.read.json("fixed.json")

三、性能最优选择 #

• 超大文件/分布式场景：优先使用Spark RDD拆分方案，利用分布式计算避免Driver内存过载，适配TB级别的数据处理。
• 中小文件：直接用Python预处理，代码简洁开销小，无需复杂的分布式流程。

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