Python JSON库json、simdjson与orjson深度对比

pip install simdjson orjson

import json
import time
import simdjson
import orjson
import random
import string


def generate_large_json(num_records: int = 50000) -> str:
    def random_string(length=10):
        return ''.join(random.choices(string.ascii_letters + string.digits, k=length))

    data = []
    for i in range(num_records):
        record = {
            "id": i,
            "name": random_string(8),
            "email": f"{random_string(5)}@example.com",
            "score": round(random.uniform(0, 100), 2),
            "tags": [random_string(4) for _ in range(random.randint(1, 5))],
            "active": random.choice([True, False])
        }
        data.append(record)

    return json.dumps(data, ensure_ascii=False)


def benchmark_pure_parse():
    print("正在生成测试数据...")
    json_str = generate_large_json(num_records=50000)
    print(f"生成完成，JSON 大小: {len(json_str) / (1024 * 1024):.2f} MB")

    # Warm-up
    small_json = '{"test": true}'
    _ = json.loads(small_json)
    _ = simdjson.Parser().parse(small_json)
    _ = orjson.loads(small_json)

    # --- 标准 json ---
    print("正在测试标准 json 库（仅解析）...")
    start = time.perf_counter()
    for _ in range(5):
        obj = json.loads(json_str)
    std_time = (time.perf_counter() - start) / 5

    # --- simdjson ---
    print("正在测试 simdjson（仅解析）...")
    start = time.perf_counter()
    for _ in range(5):
        parser = simdjson.Parser()
        obj = parser.parse(json_str)  # 注意：不调用 .as_dict()
    simd_time = (time.perf_counter() - start) / 5

    # --- orjson ---
    print("正在测试 orjson（仅解析）...")
    start = time.perf_counter()
    for _ in range(5):
        obj = orjson.loads(json_str)
    orjson_time = (time.perf_counter() - start) / 5

    # --- 结果 ---
    print("\n📊 纯解析性能对比结果:")
    print(f"标准 json 平均时间: {std_time:.6f} 秒")
    print(f"simdjson 平均时间:   {simd_time:.6f} 秒")
    print(f"orjson 平均时间:     {orjson_time:.6f} 秒")
    print(f"simdjson 加速比:     {std_time / simd_time:.2f}x")
    print(f"orjson 加速比:       {std_time / orjson_time:.2f}x")


if __name__ == "__main__":
    benchmark_pure_parse()

正在生成测试数据...
生成完成，JSON 大小: 6.25 MB
正在测试标准 json 库（仅解析）...
正在测试 simdjson（仅解析）...
正在测试 orjson（仅解析）...

📊 纯解析性能对比结果:
标准 json 平均时间: 0.124597 秒
simdjson 平均时间:   0.013677 秒
orjson 平均时间:     0.088928 秒
simdjson 加速比:     9.11x
orjson 加速比:       1.40x

import json
import time
import orjson
import simdjson
import random
import string


def generate_large_json(num_records: int = 50000) -> str:
    def random_string(length=10):
        return ''.join(random.choices(string.ascii_letters + string.digits, k=length))

    data = []
    for i in range(num_records):
        record = {
            "id": i,
            "name": random_string(8),
            "email": f"{random_string(5)}@example.com",
            "score": round(random.uniform(0, 100), 2),
            "tags": [random_string(4) for _ in range(random.randint(1, 5))],
            "active": random.choice([True, False])
        }
        data.append(record)

    return json.dumps(data, ensure_ascii=False)


def benchmark_etl():
    print("正在生成测试数据...")
    json_str = generate_large_json(num_records=50000)
    print(f"生成完成，JSON 大小: {len(json_str) / (1024 * 1024):.2f} MB")

    # Warm-up
    small_json = '{"test": true}'
    _ = json.loads(small_json)
    _ = orjson.loads(small_json)
    _ = simdjson.Parser().parse(small_json)

    # --- 标准 json：解析 → 修改 → 序列化 ---
    print("正在测试标准 json 库（完整流程）...")
    start = time.perf_counter()
    for _ in range(5):
        data = json.loads(json_str)
        for item in data:
            item["grade"] = "A" if item["score"] >= 90 else "B"
        new_json = json.dumps(data, ensure_ascii=False)
    std_time = (time.perf_counter() - start) / 5

    # --- orjson：解析 → 修改 → 序列化 ---
    print("正在测试 orjson（完整流程）...")
    start = time.perf_counter()
    for _ in range(5):
        data = orjson.loads(json_str)
        for item in data:
            item["grade"] = "A" if item["score"] >= 90 else "B"
        new_json = orjson.dumps(data)
    orjson_time = (time.perf_counter() - start) / 5

    # --- simdjson：解析 → 转 dict → 修改 → 序列化 ---
    print("正在测试 simdjson（完整流程，含 .as_list()）...")
    start = time.perf_counter()
    for _ in range(5):
        parser = simdjson.Parser()
        obj = parser.parse(json_str)
        data = obj.as_list()  # 转为可修改的 Python 对象
        for item in data:
            item["grade"] = "A" if item["score"] >= 90 else "B"
        new_json = json.dumps(data, ensure_ascii=False)
    simd_time = (time.perf_counter() - start) / 5

    # --- 结果 ---
    print("\n📊 完整 ETL 流程性能对比结果（解析 + 修改 + 序列化）:")
    print(f"标准 json 平均时间: {std_time:.6f} 秒")
    print(f"orjson 平均时间:     {orjson_time:.6f} 秒")
    print(f"simdjson 平均时间:   {simd_time:.6f} 秒")
    print(f"orjson 相对加速比:   {std_time / orjson_time:.2f}x")
    print(f"simdjson 相对加速比: {std_time / simd_time:.2f}x (越小越慢)")


if __name__ == "__main__":
    benchmark_etl()

正在生成测试数据...
生成完成，JSON 大小: 6.25 MB
正在测试标准 json 库（完整流程）...
正在测试 orjson（完整流程）...
正在测试 simdjson（完整流程，含 .as_list()）...

📊 完整 ETL 流程性能对比结果（解析 + 修改 + 序列化）:
标准 json 平均时间: 0.259784 秒
orjson 平均时间:     0.119655 秒
simdjson 平均时间:   0.328153 秒
orjson 相对加速比:   2.17x
simdjson 相对加速比: 0.79x (越小越慢)

parser = simdjson.Parser()
obj1 = parser.parse(json1)  # OK
obj2 = parser.parse(json2)  # ❌ RuntimeError!