高效处理大 JSON：四舍五入 x 并聚合 y 均值

本文介绍一种内存友好、逻辑清晰的 Python 方法，用于解析大型 JSON 文件：将 x 字段四舍五入为整数，并对相同 x 值对应的 y 值求平均，最终生成结构精简的新 JSON。适用于传感器数据、时间序列聚合等场景。

本文介绍一种内存友好、逻辑清晰的 Python 方法，用于解析大型 JSON 文件：将 x 字段四舍五入为整数，并对相同 x 值对应的 y 值求平均，最终生成结构精简的新 JSON。适用于传感器数据、时间序列聚合等场景。

在处理大规模 JSON 数据（如数万条观测记录）时，直接加载全部内容到内存可能引发性能瓶颈。但本任务的核心逻辑——x 四舍五入 → 分组 → y 均值聚合——无需全量驻留，可通过流式分块或单次遍历高效完成。以下提供一个健壮、可扩展的实现方案。

核心思路：一次遍历 + 分组字典聚合

我们不预先收集所有 x 值再去重，而是使用 defaultdict(list) 直接按四舍五入后的 x 分组存储原始 y 值。这样既避免重复扫描，又天然支持后续均值计算：

import json
from collections import defaultdict
from pathlib import Path

def process_large_json(input_path: str, output_path: str, key_field: str = "DATA1") -> None:
    """
    解析大型 JSON 文件：对 key_field 下的每个对象，
    将 'x' 四舍五入为整数，对相同 x 的 'y' 取平均值，保留 'z'（若存在且为 null 则保持）。
    """
    # 1. 读取并解析 JSON（适用于文件可一次性加载的中等规模；超大文件建议用 ijson 流式解析）
    with open(input_path, "r", encoding="utf-8") as f:
        data = json.load(f)

    if key_field not in data:
        raise KeyError(f"Key '{key_field}' not found in JSON root.")

    raw_entries = data[key_field]

    # 2. 按 round(x) 分组：key=round(x), value=list of y values
    grouped_y = defaultdict(list)
    # 同时记录每个 group 对应的 z（假设同 x 组内 z 一致或均为 null；否则需明确策略）
    group_z = {}

    for entry in raw_entries:
        x_val = entry.get("x")
        y_val = entry.get("y")
        z_val = entry.get("z")

        if x_val is None or y_val is None:
            continue  # 跳过缺失关键字段的条目

        rounded_x = round(float(x_val))  # 确保数值类型，兼容字符串输入

        grouped_y[rounded_x].append(float(y_val))
        # 若 z 非 null 且未设置，则记录；优先保留首个非 null z（可根据业务调整）
        if rounded_x not in group_z and z_val is not None:
            group_z[rounded_x] = z_val
        elif rounded_x not in group_z:
            group_z[rounded_x] = z_val  # 仍设为 None

    # 3. 构建新 DATA1 列表
    new_data1 = []
    for rx in sorted(grouped_y.keys()):  # 按 x 升序排列，提升可读性
        avg_y = sum(grouped_y[rx]) / len(grouped_y[rx])
        # 保持与原格式一致：x 和 y 为 float 类型（如 1.0），z 保持原值
        new_entry = {
            "x": float(rx),
            "y": round(avg_y, 6),  # 保留6位小数防止浮点误差过度传播
            "z": group_z.get(rx)
        }
        new_data1.append(new_entry)

    # 4. 写入新 JSON
    result = {key_field: new_data1}
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(result, f, indent=2, ensure_ascii=False)

    print(f"✅ Processed {len(raw_entries)} entries → {len(new_data1)} aggregated entries.")

# 使用示例
# process_large_json("original.json", "new.json")

关键注意事项

• 内存优化提示：若 JSON 文件远超内存（如 >500MB），请改用 ijson 库进行迭代解析，仅提取 DATA1 数组中的对象，避免一次性加载整个文档。
• 空值与异常处理：代码已跳过 x 或 y 缺失的条目；生产环境建议添加日志记录告警。
• z 字段策略：当前默认保留首个出现的非 null z 值。若需严格一致性（如所有 z 必须相同），可在分组时校验并抛出异常。
• 精度控制：round(y, 6) 防止浮点累加误差；如需更高精度，可改用 decimal.Decimal。
• 扩展性：函数支持自定义主键字段（如 "DATA2"），便于复用。

该方法时间复杂度为 O(n)，空间复杂度为 O(k)（k 为唯一 round(x) 的数量），兼顾效率与可维护性，是工业级数据预处理的推荐实践。