Python 处理 JSONL：读取、写入与解析

使用 Python 处理 JSONL（JSON Lines）文件的完整指南。学习使用内置模块、pandas 和高性能库来读取、写入、解析和流式处理 JSONL 数据。

最后更新：2026年2月

为什么选择 Python 处理 JSONL？

Python 是处理 JSONL 文件最流行的语言，这是有充分理由的。它的内置 json 模块开箱即用地支持 JSON 解析，文件迭代默认就是内存高效的，而且生态系统提供了强大的库如 pandas 和 orjson 用于专业工作流。无论您是在处理机器学习数据集、应用日志还是 API 响应，Python 都能让 JSONL 处理变得简单直接。

JSONL（JSON Lines）每行存储一个 JSON 对象，使其非常适合流式处理、追加式日志记录以及处理大型数据集而无需将所有内容加载到内存中。Python 的逐行文件读取与这种格式完美契合。在本指南中，您将学习三种读取 JSONL 的方法、两种写入方法，以及如何处理无法完全加载到内存中的大文件。

在 Python 中读取 JSONL 文件

Python 中有多种读取 JSONL 文件的方法，每种方法适合不同的使用场景。标准 json 模块适用于大多数场景，pandas 方便进行表格分析，生成器则最适合处理大文件。

最简单的方法是使用 Python 内置的 json 模块。打开文件，逐行迭代，使用 json.loads() 解析每一行。这会将所有记录加载到内存中的列表中。

使用 json 模块基础读取
import json


records = []
with open('data.jsonl', 'r', encoding='utf-8') as f:
    for line in f:
        line = line.strip()
        if line:  # Skip empty lines
            records.append(json.loads(line))


print(f'Loaded {len(records)} records')
print(records[0])

如果您的 JSONL 数据是表格形式的（每条记录有相同的键），pandas 可以通过一个函数调用直接将其读入 DataFrame。这是开始分析结构化 JSONL 数据最快的方式。

使用 pandas 读取
import pandas as pd


# Read entire file into a DataFrame
df = pd.read_json('data.jsonl', lines=True)
print(df.head())
print(f'Shape: {df.shape}')


# For large files, read in chunks
chunks = pd.read_json('large.jsonl', lines=True, chunksize=10000)
for chunk in chunks:
    # Process each chunk (DataFrame)
    print(f'Chunk shape: {chunk.shape}')

对于无法完全加载到内存中的文件，使用生成器函数。它每次只产出一条记录，无论文件大小如何都能保持恒定的内存使用量。这是生产环境数据管道的推荐模式。

大文件的生成器模式
import json
from typing import Iterator, Any


def read_jsonl(path: str) -> Iterator[dict[str, Any]]:
    """Read a JSONL file lazily, yielding one record at a time."""
    with open(path, 'r', encoding='utf-8') as f:
        for line_num, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            try:
                yield json.loads(line)
            except json.JSONDecodeError as e:
                print(f'Skipping invalid JSON at line {line_num}: {e}')


# Process records one at a time
for record in read_jsonl('large_data.jsonl'):
    process(record)  # Only one record in memory at a time

在 Python 中写入 JSONL 文件

写入 JSONL 文件非常简单：将每条记录序列化为 JSON 字符串并追加换行符。关键规则是每行一个 JSON 对象，没有尾随逗号或外围数组。

使用 json.dumps() 序列化每条记录，然后写入并追加换行符。设置 ensure_ascii=False 以在输出中保留中文、日文或 emoji 等 Unicode 字符。

使用 json 模块基础写入
import json


records = [
    {"id": 1, "name": "Alice", "age": 30},
    {"id": 2, "name": "Bob", "age": 25},
    {"id": 3, "name": "Charlie", "age": 35},
]


with open('output.jsonl', 'w', encoding='utf-8') as f:
    for record in records:
        f.write(json.dumps(record, ensure_ascii=False) + '\n')


print(f'Wrote {len(records)} records to output.jsonl')

如果您的数据已经在 pandas DataFrame 中，使用 to_json() 并设置 orient='records' 和 lines=True 即可直接导出为 JSONL。这是 pd.read_json() with lines=True 的逆操作。

使用 pandas 写入
import pandas as pd


df = pd.DataFrame([
    {"id": 1, "name": "Alice", "age": 30},
    {"id": 2, "name": "Bob", "age": 25},
    {"id": 3, "name": "Charlie", "age": 35},
])


# Write DataFrame to JSONL
df.to_json('output.jsonl', orient='records', lines=True, force_ascii=False)


print(f'Wrote {len(df)} records to output.jsonl')

Python JSONL 库

Python 提供了多个具有不同性能特征的 JSON 解析库。选择哪个取决于您的文件大小和性能需求。

json（标准库）

内置

Python 内置的 json 模块无需安装，随处可用。它能满足大多数 JSONL 工作负载，支持所有标准 JSON 类型。对于几百 MB 以内的文件，性能完全够用。

orjson

最快

orjson 是最快的 Python JSON 库，使用 Rust 编写。相比标准 json 模块，它提供 2-10 倍的解析和序列化速度提升。它输出 bytes 而非字符串，并原生支持 dataclasses、datetime、numpy 和 UUID 类型。

ujson

快速

ujson（UltraJSON）是一个基于 C 的 JSON 库，比标准 json 模块快 2-5 倍。它的 API 与内置 json 模块几乎相同，可以作为直接替换。是兼容性和速度之间的良好折中。

流式处理大型 JSONL 文件

处理 GB 级别的 JSONL 文件时，您需要一种流式方法来分批读取、转换和写入数据。这可以保持恒定的内存使用量并提供进度跟踪。

流式处理大型 JSONL 文件
import json
import sys


def process_large_jsonl(
    input_path: str,
    output_path: str,
    batch_size: int = 1000
) -> int:
    """Stream-process a large JSONL file in batches."""
    processed = 0
    batch: list[dict] = []


    with open(input_path, 'r') as fin, \
         open(output_path, 'w') as fout:
        for line in fin:
            line = line.strip()
            if not line:
                continue
            record = json.loads(line)
            # Transform the record
            record['processed'] = True
            batch.append(record)


            if len(batch) >= batch_size:
                for r in batch:
                    fout.write(json.dumps(r) + '\n')
                processed += len(batch)
                batch.clear()
                print(f'\rProcessed {processed} records...', end='')


        # Write remaining records
        for r in batch:
            fout.write(json.dumps(r) + '\n')
        processed += len(batch)


    print(f'\nDone. Processed {processed} records total.')
    return processed


# Usage
process_large_jsonl('input.jsonl', 'output.jsonl', batch_size=5000)

此模式通过固定大小的批次处理记录来实现恒定内存使用。batch_size 参数控制内存使用与 I/O 效率之间的权衡。对于大多数系统，1,000 到 10,000 条记录的批次效果良好。进度指示器有助于监控长时间运行的任务。

试试我们的免费 JSONL 工具

不想写代码？使用我们的免费在线工具，直接在浏览器中查看、验证和转换 JSONL 文件。

large JSONL files guide
JSONL validator
JSONL best practices

在线处理 JSONL 文件

直接在浏览器中查看、验证和转换最大 1GB 的 JSONL 文件。无需上传，100% 私密。

常见问题

Python 处理 JSONL — 读取、写入、流式处理与验证 JSON Lines | jsonl.co