如何在Python中高效处理内存受限下的百万级数据集过滤与加工？

嘿，我之前也踩过百万级数据集内存溢出的坑，你的需求太典型了——全量数据塞不下内存，还要做条件过滤和字段加工。咱们来一步步优化你的方案，彻底解决内存问题：

核心问题分析 #

你原来的代码把所有过滤加工后的结果都存在filtered_and_processed列表里，百万条记录的话这个列表会直接占满内存。而且如果你的原始数据是从文件/数据库加载的，一次性读成列表本身就会先爆内存。解决的关键就是用惰性求值代替一次性存储，让数据“流”起来。

方案1：用生成器实现流式处理 #

生成器是Python处理大内存数据的神器，它不会一次性把所有结果存在内存里，而是每次迭代才生成一条记录，内存占用几乎恒定。

改写你的代码：

def process_record(record):
    # 建议创建新字典返回，避免修改原数据产生副作用
    return {
        'name': record['name'].lower(),
        'email_domain': record['email'].split('@')[1]
    }

def filter_and_process(data_iter):
    # 接收一个迭代器作为输入，惰性处理每条记录
    for record in data_iter:
        if record['age'] > 25 and record['status'] == 'active':
            yield process_record(record)

# 测试用你的示例数据
data = [
    {'name': 'Alice', 'age': 30, 'status': 'active', 'email': 'alice@example.com'},
    {'name': 'Bob', 'age': 22, 'status': 'inactive', 'email': 'bob@example.com'},
    {'name': 'Charlie', 'age': 35, 'status': 'active', 'email': 'charlie@example.com'},
    # 更多记录...
]

# 生成器不会立即处理数据，只有迭代的时候才会干活
results = filter_and_process(data)

# 遍历结果，每次只加载一条到内存
for res in results:
    print(res)
    # 这里可以直接把结果写入文件/数据库，不用存到列表里

方案2：从源头避免加载全量数据 #

如果你的原始数据是存储在文件（CSV/JSON）或者数据库里，千万不要一次性把所有数据读成列表！要做成迭代器逐行读取：

示例：处理CSV格式的大文件

import csv

def filter_and_process_csv(file_path):
    with open(file_path, 'r') as f:
        # csv.DictReader本身就是迭代器，逐行读数据
        reader = csv.DictReader(f)
        for row in reader:
            # CSV读出来的都是字符串，记得转类型
            age = int(row['age'])
            if age > 25 and row['status'] == 'active':
                yield {
                    'name': row['name'].lower(),
                    'email_domain': row['email'].split('@')[1]
                }

# 使用方式：直接遍历生成器，边处理边输出
for processed in filter_and_process_csv('large_dataset.csv'):
    # 比如写入结果文件
    with open('processed_results.csv', 'a') as out_f:
        out_f.write(f"{processed['name']},{processed['email_domain']}\n")

示例：处理JSON Lines格式的大文件

如果你的数据是每行一个JSON对象（这种格式特别适合大文件），可以逐行解析：

import json

def filter_and_process_jsonl(file_path):
    with open(file_path, 'r') as f:
        for line in f:
            record = json.loads(line.strip())
            if record['age'] > 25 and record['status'] == 'active':
                yield process_record(record)

方案3：进阶用Pandas分块处理（适合内存不是极端紧张的情况） #

如果你习惯用Pandas做数据处理，可以用分块读取的方式，每次只加载一小部分数据到内存：

import pandas as pd

# 每次处理10000条，可根据内存情况调整
chunk_size = 10000

# 分块读取源文件
for chunk in pd.read_csv('large_dataset.csv', chunksize=chunk_size):
    # 条件过滤
    filtered_chunk = chunk[(chunk['age'] > 25) & (chunk['status'] == 'active')]
    # 字段加工
    filtered_chunk['name'] = filtered_chunk['name'].str.lower()
    filtered_chunk['email_domain'] = filtered_chunk['email'].str.split('@').str[1]
    # 把结果追加写入文件，不用存到内存里
    filtered_chunk.to_csv('processed_results.csv', mode='a', header=False, index=False)

关键总结 #

不管用哪种方案，核心思路都是：

• 不要一次性加载全量数据到内存
• 不要把所有处理后的结果存在列表里
• 让数据流式流动：读一条→过滤→加工→输出（持久化）→丢弃，循环往复

备注：内容来源于stack exchange，提问作者user20603914