在实际数据处理场景中,我们经常会遇到体积较大的JSON文件,直接读取处理会占用大量内存,甚至导致程序崩溃。使用Python将JSON文件分割成多个小文件是一种高效的解决方案,既能降低内存压力,也方便后续分布式处理或分批次操作数据。

基础准备
首先需要确保Python环境已安装,标准库中的json模块已经可以满足大部分JSON处理需求,不需要额外安装第三方库。我们需要先了解JSON文件的常见结构,通常分为两种:一种是JSON数组格式,即整个文件是一个数组,数组中的每个元素是一条独立数据;另一种是每行一个JSON对象的格式,也就是JSON Lines格式。
按数据条数分割JSON文件
最常见的分割需求是按照指定的数据条数,将原始JSON文件拆分成多个包含固定数量数据的小文件。以下代码实现了将JSON数组文件按每N条数据分割成一个新文件的功能。
import json
import os
def split_json_by_count(input_path, output_dir, count_per_file):
# 读取原始JSON文件
with open(input_path, 'r', encoding='utf-8') as f:
data = json.load(f)
# 校验数据是否为列表格式
if not isinstance(data, list):
raise ValueError("输入的JSON文件内容不是数组格式,无法按条数分割")
# 创建输出目录
os.makedirs(output_dir, exist_ok=True)
# 计算需要生成多少个文件
total_count = len(data)
file_num = 0
# 循环拆分数据
for i in range(0, total_count, count_per_file):
file_num += 1
# 截取当前文件对应的数据片段
chunk = data[i:i+count_per_file]
# 生成输出文件路径
output_path = os.path.join(output_dir, f"split_{file_num}.json")
# 写入新文件
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(chunk, f, ensure_ascii=False, indent=2)
print(f"已生成文件:{output_path},包含{len(chunk)}条数据")
print(f"分割完成,共生成{file_num}个文件")
# 调用示例
if __name__ == "__main__":
split_json_by_count("input.json", "output_json", 100)
代码说明
上述代码首先读取输入的JSON文件,判断内容是否为数组格式,避免非预期的数据结构导致错误。然后创建输出目录,按照指定的每文件数据条数循环截取数据片段,每个片段写入一个独立的JSON文件,文件名按序号递增命名。使用ensure_ascii=False可以保证中文正常显示,indent=2可以让输出的JSON文件有缩进,可读性更强。
按文件大小分割JSON文件
如果需要按照每个输出文件的大小(比如每个文件不超过1MB)来分割,可以先计算每个数据项的大致大小,再动态调整每个文件包含的数据条数。以下是实现按大小分割的示例代码。
import json
import os
def split_json_by_size(input_path, output_dir, max_size_kb):
# 读取原始JSON数据
with open(input_path, 'r', encoding='utf-8') as f:
data = json.load(f)
if not isinstance(data, list):
raise ValueError("输入的JSON文件内容不是数组格式,无法按大小分割")
os.makedirs(output_dir, exist_ok=True)
current_chunk = []
file_num = 0
# 遍历所有数据项
for item in data:
# 将当前项加入临时片段
current_chunk.append(item)
# 计算当前片段的大小(单位KB)
chunk_size = len(json.dumps(current_chunk, ensure_ascii=False).encode('utf-8')) / 1024
# 如果超过最大大小,写入文件并重置临时片段
if chunk_size >= max_size_kb:
file_num += 1
output_path = os.path.join(output_dir, f"split_{file_num}.json")
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(current_chunk, f, ensure_ascii=False, indent=2)
print(f"已生成文件:{output_path},大小约{chunk_size:.2f}KB")
current_chunk = []
# 处理最后剩余的数据
if current_chunk:
file_num += 1
output_path = os.path.join(output_dir, f"split_{file_num}.json")
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(current_chunk, f, ensure_ascii=False, indent=2)
chunk_size = len(json.dumps(current_chunk, ensure_ascii=False).encode('utf-8')) / 1024
print(f"已生成文件:{output_path},大小约{chunk_size:.2f}KB")
print(f"分割完成,共生成{file_num}个文件")
# 调用示例,每个文件最大100KB
if __name__ == "__main__":
split_json_by_size("input.json", "output_json", 100)
处理JSON Lines格式文件
如果原始文件是每行一个JSON对象的JSON Lines格式,分割逻辑会更简单,不需要整体解析JSON数组,直接按行读取拆分即可。
import os
def split_jsonlines_by_count(input_path, output_dir, count_per_file):
os.makedirs(output_dir, exist_ok=True)
file_num = 0
current_count = 0
output_file = None
# 逐行读取原始文件
with open(input_path, 'r', encoding='utf-8') as f:
for line in f:
# 去除空行
line = line.strip()
if not line:
continue
# 如果当前文件达到条数限制,关闭旧文件,创建新文件
if current_count % count_per_file == 0:
if output_file:
output_file.close()
file_num += 1
output_path = os.path.join(output_dir, f"split_{file_num}.jsonl")
output_file = open(output_path, 'w', encoding='utf-8')
print(f"开始写入文件:{output_path}")
# 写入当前行
output_file.write(line + "n")
current_count += 1
if output_file:
output_file.close()
print(f"分割完成,共生成{file_num}个JSON Lines文件")
# 调用示例
if __name__ == "__main__":
split_jsonlines_by_count("input.jsonl", "output_jsonlines", 100)
常见问题与注意事项
- 分割前建议先备份原始JSON文件,避免操作失误导致数据丢失。
- 如果JSON文件结构不是数组格式,而是嵌套的复杂对象,需要先明确拆分规则,比如按照某个字段的值拆分,再调整代码逻辑。
- 处理超大JSON文件时,如果一次性读取到内存会导致内存不足,可以使用
ijson等流式解析库逐部分读取数据,再进行分割。 - 输出文件的编码建议统一使用utf-8,避免中文乱码问题。
PythonJSON文件分割json_split文件处理修改时间:2026-07-16 19:27:27