You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在Python中高效移除超大规模列表中的重复项?

针对超大规模列表去重的高效方案

处理1亿+元素的列表去重,直接用Set全量加载确实会因为内存占用过高导致性能瓶颈,以下是几个更高效的解决方案:

1. 分块处理+外部存储

把大列表拆分成若干小分块,每次只加载一个分块到内存,用Set去重后写入临时文件;所有分块处理完成后,再逐一读取临时文件内容,合并成最终的去重结果。这种方式将内存压力分散到多次处理,避免一次性加载所有数据。

示例代码(Python):

import os

def chunked_deduplicate(input_list, chunk_size=1000000):
    temp_files = []
    # 分块处理并写入临时文件
    for idx in range(0, len(input_list), chunk_size):
        chunk = input_list[idx:idx+chunk_size]
        dedup_chunk = list(set(chunk))
        temp_file_path = f"temp_dedup_{idx//chunk_size}.tmp"
        with open(temp_file_path, 'w', encoding='utf-8') as f:
            f.write('\n'.join(map(str, dedup_chunk)))
        temp_files.append(temp_file_path)
    
    # 合并临时文件并生成最终结果
    final_dedup = set()
    for file_path in temp_files:
        with open(file_path, 'r', encoding='utf-8') as f:
            for line in f:
                item = line.strip()
                final_dedup.add(item)
        os.remove(file_path)  # 清理临时文件
    return list(final_dedup)

2. 排序后遍历去重

先对列表进行排序(内存不足时使用外部排序,比如系统自带的sort命令),排序后重复元素会相邻,只需遍历一次即可完成去重,无需占用额外的内存存储整个哈希集合。排序的时间复杂度为O(n log n),但内存占用远低于全量Set方案。

  • 若数据能放入内存,直接用内存排序:
def sorted_deduplicate(input_list):
    sorted_items = sorted(input_list)
    dedup_result = []
    prev_item = None
    for item in sorted_items:
        if item != prev_item:
            dedup_result.append(item)
            prev_item = item
    return dedup_result
  • 若数据无法放入内存,用系统工具处理(Linux/macOS):
# 假设原始数据存在large_list.txt中,每行一个元素
sort large_list.txt | uniq > deduped_list.txt

系统的sort命令默认支持外部排序,会自动利用磁盘存储中间结果,效率极高。

3. 哈希分桶去重

根据元素的哈希值将整个列表拆分成多个“桶”,每个桶中的元素哈希值模后结果相同,重复元素一定会被分到同一个桶中。逐个处理每个桶的去重(每个桶的数据量远小于原列表,可轻松放入内存),最后合并所有桶的结果即可。

示例代码(Python):

def hash_bucket_deduplicate(input_list, bucket_count=100):
    # 初始化桶
    buckets = [[] for _ in range(bucket_count)]
    # 分桶
    for item in input_list:
        bucket_idx = hash(item) % bucket_count
        buckets[bucket_idx].append(item)
    # 每个桶单独去重后合并
    dedup_result = []
    for bucket in buckets:
        dedup_result.extend(list(set(bucket)))
    return dedup_result

注意:若存在哈希碰撞(不同元素哈希值相同),不会影响去重结果,因为同一元素的哈希值必然相同,会被分到同一桶中去重。

4. 利用数据库去重

将数据导入数据库,利用数据库的唯一约束或DISTINCT关键字自动去重,无需自己实现复杂的分块或排序逻辑。适合不擅长编写底层处理代码的场景。

示例代码(SQLite):

import sqlite3
import os

def db_deduplicate(input_list):
    # 用磁盘数据库而非内存数据库,避免内存溢出
    conn = sqlite3.connect('dedup_db.sqlite')
    cursor = conn.cursor()
    # 创建带唯一约束的表
    cursor.execute('CREATE TABLE IF NOT EXISTS items (value TEXT UNIQUE)')
    # 批量插入,忽略重复项
    try:
        cursor.executemany('INSERT OR IGNORE INTO items VALUES (?)', [(str(item),) for item in input_list])
    except Exception as e:
        print(f"插入异常: {e}")
    conn.commit()
    # 查询去重结果
    cursor.execute('SELECT value FROM items')
    dedup_result = [row[0] for row in cursor.fetchall()]
    # 清理数据库文件(可选)
    conn.close()
    os.remove('dedup_db.sqlite')
    return dedup_result

内容的提问来源于stack exchange,提问作者Harmony Chikari

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.05 09:31:42