如何快速解析大型JSON嵌套列表并优化数据处理性能？

问题场景 #

通过多线程调用数千次分页API，得到嵌套JSON列表（每个内层列表对应一页数据），每页最多1000条宠物数据，最终要生成约8万行CSV。当前代码因嵌套循环去重导致运行极慢，且重复ID可能出现在任意位置，无法提前去重。

当前代码的核心性能瓶颈 #

• 双重循环去重的O(n²)复杂度：原代码用两层for循环对比所有ID，8万条数据会产生64亿次对比操作，这是速度慢的核心原因。
• 重复索引访问：多次重复访问all_pet_data[p]['leaderboard'][i]，每次索引都有额外开销。
• 冗余的数据处理步骤：多次对csv_data做筛选，产生不必要的内存拷贝。
• 重复的错误检查：两次try-except做相同的leaderboard长度检查，代码冗余。

优化方案 #

1. 用集合实现O(1)时间复杂度的去重 #

用Python的set存储已出现的ID，每次解析数据前先检查ID是否在集合中，不在则保留数据并加入集合，彻底替代低效的双重循环。

2. 缓存每页的leaderboard数据 #

每次循环时先把all_pet_data[p]['leaderboard']存到变量里，避免重复索引嵌套字典，减少不必要的开销。

3. 合并错误处理逻辑 #

把两次检查leaderboard的try-except合并成一次，精简冗余代码。

4. 边解析边处理，减少中间列表的内存占用 #

不需要先把所有数据存入csv_data再去重，而是解析一条、检查一条、合格就加入结果列表，同时维护去重集合，减少内存占用和后续处理步骤。

5. 可选：直接写入CSV（内存敏感场景） #

如果数据量极大，可以不保存完整结果列表，直接在解析时写入CSV文件，进一步降低内存压力。

优化后的完整代码 #

import csv
from datetime import datetime

today_date = datetime.today().strftime("%Y%m%d")  # 补充原代码缺失的日期变量定义

def save(csv_data):
    with open(f"{today_date}.csv", "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerows(csv_data)

# 初始化变量
error_count = 0
pet_count = 0
seen_ids = set()  # 用于快速去重的集合
csv_data = []

# 直接遍历分页数据，避免索引循环
for page_data in all_pet_data:
    try:
        leaderboard = page_data['leaderboard']
        current_page_count = len(leaderboard)
        pet_count += current_page_count
        
        # 直接遍历宠物数据，避免索引循环
        for pet in leaderboard:
            pet_id = pet['id']
            if pet_id not in seen_ids:
                seen_ids.add(pet_id)
                csv_data.append([pet_id, pet['level'], pet['name']])
    except Exception as e:
        print(f'{e} when processing page data: {page_data}')
        error_count += 1

# 过滤空行（边解析边处理已大幅减少这类情况）
csv_data = [row for row in csv_data if row]

# 调试输出
print(f'error_count: {error_count}')
print(f'Total pets grabbed: {pet_count}')
print(f'Total unique pets grabbed: {len(csv_data)}')

save(csv_data)

内存敏感场景的额外优化 #

如果数据量超过内存承载能力，可以直接在解析时写入CSV，不需要保存完整的csv_data列表：

def save_directly():
    with open(f"{today_date}.csv", "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        for page_data in all_pet_data:
            try:
                leaderboard = page_data['leaderboard']
                global pet_count, error_count
                pet_count += len(leaderboard)
                for pet in leaderboard:
                    pet_id = pet['id']
                    if pet_id not in seen_ids:
                        seen_ids.add(pet_id)
                        writer.writerow([pet_id, pet['level'], pet['name']])
            except Exception as e:
                print(f'{e} when processing page data: {page_data}')
                error_count += 1

# 初始化全局变量后调用该函数
seen_ids = set()
pet_count = 0
error_count = 0
save_directly()

# 调试输出
print(f'error_count: {error_count}')
print(f'Total pets grabbed: {pet_count}')
print(f'Total unique pets grabbed: {len(seen_ids)}')

内容的提问来源于stack exchange，提问作者Brian - RGY Studio