本文介绍如何在不加载全部数据到内存的前提下，将大体积（如 1.1 GB）的“每行一个 JSON 对象”文件（JSON Lines 格式）转换为符合 RFC 8259 的单个有效 JSON 文件，包含外层 Records 数组封装，并支持批量处理多个文件。
本文介绍如何在不加载全部数据到内存的前提下，将大体积（如 1.1 gb）的“每行一个 json 对象”文件（json lines 格式）转换为符合 rfc 8259 的单个有效 json 文件，包含外层 `records` 数组封装，并支持批量处理多个 files。
在实际数据工程中，常遇到以 JSON Lines（NDJSON） 格式存储的日志或导出数据：每个换行符分隔一个独立、合法的 JSON 对象，例如：
{"job": "developer"}
{"job": "taxi driver"}
{"job": "police"}这类格式便于流式写入和追加，但不符合标准 JSON 规范（标准 JSON 要求且仅允许一个顶层值），因此无法被 json.load() 直接解析，也不适用于需严格校验的下游系统（如 Spark Structured Streaming、AWS Athena 或前端 JSON Schema 验证器）。目标是将其无损、高效地重构为：
{
"Records": [
{"job": "developer"},
{"job": "taxi driver"},
{"job": "police"}
]
}✅ 推荐方案：流式字符串拼接（零内存解析）
针对 1.1 GB 大文件，绝对避免 f.readlines() 或 json.loads() 全量解析——这会引发内存爆炸（Python 字典/对象开销远超原始文本）。正确做法是跳过反序列化，直接按行构造合法 JSON 文本：
def convert_json_lines_to_wrapped_array(input_path: str, output_path: str, key: str = "Records") -> None:
"""
将 JSON Lines 文件流式转换为带外层数组的 JSON 文件。
无需解析 JSON，纯文本拼接，内存占用恒定 O(1)。
"""
with open(input_path, 'r', encoding='utf-8') as fin, \
open(output_path, 'w', encoding='utf-8') as fout:
# 写入开头：{"Records":[
fout.write(f'{{"{key}":[\n')
# 逐行读取并写入（自动处理逗号分隔）
first_line = True
for line in fin:
line = line.strip()
if not line: # 跳过空行
continue
if not first_line:
fout.write(',\n')
fout.write(line)
first_line = False
# 写入结尾：]}
fout.write('\n]}')? 为什么高效？
- 每次仅读取一行（默认缓冲，内存占用 < 1 KB），不构建 Python 对象；
- 避免 json.dumps() 序列化开销（尤其是对嵌套结构）；
- 时间复杂度 O(n)，空间复杂度 O(1)。
? 批量处理多个文件
若目录 ./input/ 下有数百个 .json 文件，可结合 pathlib 安全批量转换：
from pathlib import Path
def batch_convert(input_dir: str = "input", output_dir: str = "output", key: str = "Records"):
input_path = Path(input_dir)
output_path = Path(output_dir)
output_path.mkdir(exist_ok=True) # 确保输出目录存在
for json_file in input_path.glob("*.json"):
if json_file.is_file():
out_file = output_path / json_file.name
try:
convert_json_lines_to_wrapped_array(
str(json_file), str(out_file), key
)
print(f"✅ Converted: {json_file.name}")
except Exception as e:
print(f"❌ Failed on {json_file.name}: {e}")
# 执行批量转换
batch_convert()⚠️ 注意事项与最佳实践
- 编码兼容性：显式指定 encoding='utf-8'，防止中文等非 ASCII 字符乱码；
- 空行/注释处理：原始 JSON Lines 不允许注释，但允许空行；上述代码已跳过空行；
-
验证输出：转换后建议用 jq 快速校验：
jq -n 'empty' ./output/sample.json # 无报错即为合法 JSON
- 错误容忍（进阶）：若输入可能含非法 JSON 行，可添加 try/except 并记录错误行号，而非中断整个流程；
- 替代工具链：对超大规模场景（TB 级），推荐使用 jq 命令行（jq -s '{Records: .}' *.json）或 Apache NiFi，Python 方案仍是最易集成的通用解法。
通过该方法，你可在数秒内完成 GB 级文件的合规化封装，兼顾性能、可靠性和可维护性。
