超大CSV文件列唯一值统计的内存溢出问题及解决方案

你的代码在小文件上运行正常，但处理20GB以上超大文件时出现内存不足、速度变慢的核心原因是把所有列的非空值及其计数全存在内存里（就是代码里的column_data嵌套字典）。当列的唯一值数量达到百万甚至亿级时，内存会被快速耗尽，频繁的垃圾回收拖慢速度，最终触发内存溢出错误。下面给你几个实用的解决办法：

一、用SQLite替代内存存储（精确统计） #

把唯一值的计数放到磁盘上的SQLite数据库中，内存只需要处理当前行的数据，完全不用担心内存不够。SQLite轻量高效，处理这种键值更新的性能足够支撑超大文件处理。

修改后的代码示例：

import csv
import sqlite3
from contextlib import closing

def read_large_csv(file_path, delimiter, encoding):
    # 连接磁盘上的SQLite数据库（避免内存数据库重启丢失数据）
    with closing(sqlite3.connect('csv_stats.db')) as conn:
        cursor = conn.cursor()
        # 创建表存储各列的唯一值计数，联合主键确保列+值唯一
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS column_values (
                column_name TEXT, 
                value TEXT, 
                count INTEGER, 
                PRIMARY KEY (column_name, value)
            )
        ''')
        
        with open(file_path, encoding=encoding) as f:
            reader = csv.reader(f, delimiter=delimiter)
            header = next(reader)
            # 初始化基础统计项
            columns = {
                col_name: {'total': 0, 'not_null': 0, 'unique': 0, 'first_five': set()} 
                for col_name in header
            }
            num = 0
            
            for row in reader:
                if num % 1000000 == 0:
                    print(f"{num / 1000000} M rows processed")
                num += 1
                
                for col_index, col_value in enumerate(row):
                    col_name = header[col_index]
                    columns[col_name]['total'] += 1
                    
                    if col_value:
                        columns[col_name]['not_null'] += 1
                        # 插入或更新计数：存在则+1，不存在则初始化为1
                        cursor.execute('''
                            INSERT OR REPLACE INTO column_values 
                            (column_name, value, count) 
                            VALUES (?, ?, COALESCE((SELECT count FROM column_values WHERE column_name=? AND value=?), 0) + 1)
                        ''', (col_name, col_value, col_name, col_value))
                        # 检查是否是首次出现的新值
                        cursor.execute('SELECT count FROM column_values WHERE column_name=? AND value=?', 
                                      (col_name, col_value))
                        count = cursor.fetchone()[0]
                        if count == 1:
                            columns[col_name]['unique'] += 1
                            if len(columns[col_name]['first_five']) < 5:
                                columns[col_name]['first_five'].add(col_value)
                # 每处理1000行提交一次，避免频繁IO
                if num % 1000 == 0:
                    conn.commit()
            # 提交剩余操作
            conn.commit()
            
            # 输出统计结果
            for col_name, stats in columns.items():
                print(f"列 {col_name}:")
                print(f"  总行数: {stats['total']}")
                print(f"  非空值数: {stats['not_null']}")
                print(f"  唯一值数: {stats['unique']}")
                print(f"  前5个唯一值: {stats['first_five']}")

二、用HyperLogLog做近似统计（内存占用极低） #

如果业务可以接受1%左右的误差，HyperLogLog算法是最优解。它只需要几KB到几十KB的内存，就能快速估算出百万甚至亿级别的唯一值数量，完全不会出现内存问题。

先安装依赖库：pip install hyperloglog

代码示例：

import csv
from hyperloglog import HyperLogLog

def read_large_csv_approx(file_path, delimiter, encoding):
    with open(file_path, encoding=encoding) as f:
        reader = csv.reader(f, delimiter=delimiter)
        header = next(reader)
        # 每个列初始化一个HyperLogLog实例（0.01代表误差率）
        hll_instances = {col_name: HyperLogLog(0.01) for col_name in header}
        columns = {
            col_name: {'total': 0, 'not_null': 0, 'first_five': set()} 
            for col_name in header
        }
        num = 0
        
        for row in reader:
            if num % 1000000 == 0:
                print(f"{num / 1000000} M rows processed")
            num += 1
            
            for col_index, col_value in enumerate(row):
                col_name = header[col_index]
                columns[col_name]['total'] += 1
                
                if col_value:
                    columns[col_name]['not_null'] += 1
                    # 添加值到HyperLogLog（需要转成字节）
                    hll_instances[col_name].add(col_value.encode('utf-8'))
                    # 记录前5个唯一值
                    if len(columns[col_name]['first_five']) < 5:
                        columns[col_name]['first_five'].add(col_value)
        
        # 输出近似统计结果
        for col_name, stats in columns.items():
            approx_unique = hll_instances[col_name].cardinality()
            print(f"列 {col_name}:")
            print(f"  总行数: {stats['total']}")
            print(f"  非空值数: {stats['not_null']}")
            print(f"  近似唯一值数: {approx_unique}")
            print(f"  前5个唯一值: {stats['first_five']}")

三、分块处理+合并结果（精确统计，适合200GB级文件） #

把超大文件分割成多个小文件（比如每个1GB对应约百万行），逐个处理每个小文件得到该文件的唯一值集合，最后合并所有小文件的结果去重，得到总数。这样每个小文件处理时内存可控，完全适配超大规模文件。

代码示例（分割+处理+合并）：

import csv
import os
from collections import defaultdict

def split_large_csv(file_path, chunk_size=10**6, encoding='utf-8'):
    """把大文件分割成每个chunk_size行的小文件"""
    chunk_num = 0
    with open(file_path, encoding=encoding) as f:
        reader = csv.reader(f)
        header = next(reader)
        while True:
            chunk_name = f"chunk_{chunk_num}.csv"
            with open(chunk_name, 'w', encoding=encoding, newline='') as chunk_f:
                writer = csv.writer(chunk_f)
                writer.writerow(header)
                # 写入chunk_size行数据
                for _ in range(chunk_size):
                    try:
                        row = next(reader)
                        writer.writerow(row)
                    except StopIteration:
                        chunk_num +=1
                        return chunk_num
                chunk_num +=1

def process_chunk(chunk_path, encoding='utf-8'):
    """处理单个小文件，返回该文件的统计结果"""
    with open(chunk_path, encoding=encoding) as f:
        reader = csv.reader(f)
        header = next(reader)
        chunk_stats = {
            col: {'not_null':0, 'unique': set()} 
            for col in header
        }
        total_rows = 0
        
        for row in reader:
            total_rows +=1
            for idx, val in enumerate(row):
                col_name = header[idx]
                if val:
                    chunk_stats[col_name]['not_null'] +=1
                    chunk_stats[col_name]['unique'].add(val)
        
        return total_rows, chunk_stats

def merge_results(all_chunk_results):
    """合并所有小文件的统计结果"""
    merged = {}
    total_total_rows = 0
    for total_rows, chunk_data in all_chunk_results:
        total_total_rows += total_rows
        for col_name, stats in chunk_data.items():
            if col_name not in merged:
                merged[col_name] = {'total':0, 'not_null':0, 'unique': set()}
            merged[col_name]['total'] += total_rows
            merged[col_name]['not_null'] += stats['not_null']
            merged[col_name]['unique'].update(stats['unique'])
    
    # 把唯一值集合转为数量
    for col in merged:
        merged[col]['unique'] = len(merged[col]['unique'])
    merged['total_rows'] = total_total_rows
    return merged

# 使用示例
if __name__ == "__main__":
    file_path = "huge_file.csv"
    delimiter = ","
    encoding = "utf-8"
    
    print("分割大文件...")
    chunk_count = split_large_csv(file_path, encoding=encoding)
    
    print("处理每个小文件...")
    all_results = []
    for i in range(chunk_count):
        chunk_path = f"chunk_{i}.csv"
        print(f"处理第 {i+1}/{chunk_count} 个小文件")
        res = process_chunk(chunk_path, encoding=encoding)
        all_results.append(res)
        # 处理完删除小文件节省磁盘空间
        os.remove(chunk_path)
    
    print("合并统计结果...")
    final_stats = merge_results(all_results)
    
    # 输出最终结果
    print(f"总行数: {final_stats['total_rows']}")
    for col_name, stats in final_stats.items():
        if col_name == 'total_rows':
            continue
        print(f"列 {col_name}:")
        print(f"  非空值数: {stats['not_null']}")
        print(f"  唯一值数: {stats['unique']}")

内容的提问来源于stack exchange，提问作者Myky