PySpark读取S3中无换行多JSON对象文件的问题

当你的JSON文件里多个对象是直接拼接在一起（比如{"key1":val1}{"key2":val2}），Spark默认按行解析JSON的方式就失效了——它找不到每行一个的独立JSON对象。这里给你两种实用的处理方式：

方法一：DataFrame API处理 #

先把整个文件读成单条文本记录，拆分出每个独立的JSON对象后，再转为DataFrame：

from pyspark.sql import functions as F

# 读取整个文件为单条文本记录
text_df = spark.read.text("s3a://somejsonfile", wholetext=True)

# 把}{替换成}\n{，让每个JSON对象单独成行，再拆分并展开为单独的字符串行
processed_df = text_df.withColumn(
    "json_str",
    F.explode(F.split(F.regexp_replace(F.col("value"), "}{", "}\n{"), "\n"))
).drop("value")

# 从处理后的JSON字符串生成最终DataFrame
final_df = spark.read.json(processed_df.rdd.map(lambda row: row.json_str))

final_df.show(100)

方法二：RDD API处理 #

如果习惯用RDD操作，也可以这么实现：

# 读取文件内容（wholeTextFiles返回(文件名, 文件内容)的RDD，取values只保留文件内容）
text_rdd = spark.sparkContext.wholeTextFiles("s3a://somejsonfile").values()

# 拆分每个文件里的JSON对象，得到每个对象单独的字符串
json_rdd = text_rdd.flatMap(lambda content: content.replace("}{", "}\n{").split("\n"))

# 转为DataFrame
final_df = spark.read.json(json_rdd)

final_df.show(100)

进阶注意事项 #

如果你的JSON字符串里本身包含}{（比如某个字段值里有这个字符，不过规范JSON会转义这类字符，一般不会出现），直接替换会出错。这时可以用Python的JSON解析器来精准拆分：

import json

def split_json(content):
    decoder = json.JSONDecoder()
    pos = 0
    result = []
    while pos < len(content):
        try:
            obj, pos = decoder.raw_decode(content, pos)
            result.append(json.dumps(obj))
        except json.JSONDecodeError:
            pos += 1
    return result

# 用自定义替换逻辑替换之前的flatMap操作
json_rdd = text_rdd.flatMap(split_json)

这种方法更可靠，能处理各种复杂的JSON嵌套场景，只是性能比简单替换稍慢，适合数据量不是特别大的场景。

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