如何对1500个大压缩文件计算两两交集的整数数量？

核心痛点拆解 #

你的问题核心是磁盘IO瓶颈（加载压缩文件耗时）和内存限制（16GB无法同时加载大量文件），同时需要高效完成数百万次两两交集计算。以下是针对性的Python方案：

第一步：预处理优化（解决加载慢问题） #

原始压缩文件（尤其是文本压缩）加载解压耗时，必须先转成快速读取、低内存占用的二进制格式，同时完成去重+排序（为后续高效交集计算打基础）。

预处理步骤 #

1. 读取压缩文件：根据压缩格式（gzip/xz/bzip2）用对应模块读取整数。
2. 去重排序：用numpy.unique一次性完成去重和排序（比单独排序+去重效率高）。
3. 保存为.npy格式：numpy的二进制格式支持内存映射，加载速度比压缩文本快10~100倍，且无需全量加载到内存。

预处理代码示例 #

import numpy as np
import gzip
import os

def preprocess_compressed_file(input_path, output_path):
    # 读取gzip压缩的文本整数文件（每行一个整数）
    with gzip.open(input_path, 'rt') as f:
        # 批量读取并转成int32数组（内存占用仅为int64的一半）
        arr = np.array([int(line.strip()) for line in f], dtype=np.int32)
    # 去重+排序，得到唯一整数数组
    unique_sorted_arr = np.unique(arr)
    # 保存为.npy格式（禁止pickle，更安全）
    np.save(output_path, unique_sorted_arr, allow_pickle=False)

# 批量处理所有文件
input_dir = "/path/to/your/compressed/files"
output_dir = "/path/to/preprocessed/npy_files"
os.makedirs(output_dir, exist_ok=True)

for filename in os.listdir(input_dir):
    if filename.endswith(".gz"):  # 根据你的压缩格式调整
        input_path = os.path.join(input_dir, filename)
        output_filename = f"{os.path.splitext(filename)[0]}.npy"
        output_path = os.path.join(output_dir, output_filename)
        preprocess_compressed_file(input_path, output_path)

第二步：高效交集计算（解决内存与速度问题） #

预处理后的数组是排序去重的，用双指针法计算交集数量，无需将整个数组加载到内存（利用numpy的内存映射），时间复杂度O(n+m)，比集合交集省内存且更快。

核心计算逻辑 #

def count_intersection(arr_x, arr_y):
    count = 0
    i = j = 0
    len_x, len_y = len(arr_x), len(arr_y)
    while i < len_x and j < len_y:
        if arr_x[i] == arr_y[j]:
            count += 1
            i += 1
            j += 1
        elif arr_x[i] < arr_y[j]:
            i += 1
        else:
            j += 1
    return count

并行批量计算（利用多核CPU） #

两两交集计算是CPU密集型任务，用多进程并行处理（避开GIL限制），同时控制进程数避免磁盘IO过载：

import numpy as np
import os
from concurrent.futures import ProcessPoolExecutor

def process_file_pair(pair_data):
    file_x, file_y, npy_dir = pair_data
    # 内存映射加载.npy文件（仅在需要时读取磁盘，不占满内存）
    arr_x = np.load(os.path.join(npy_dir, file_x), mmap_mode='r')
    arr_y = np.load(os.path.join(npy_dir, file_y), mmap_mode='r')
    count = count_intersection(arr_x, arr_y)
    # 手动释放内存（避免多进程内存泄漏）
    del arr_x, arr_y
    return (file_x, file_y, count)

def main():
    npy_dir = "/path/to/preprocessed/npy_files"
    file_list = [f for f in os.listdir(npy_dir) if f.endswith(".npy")]
    num_files = len(file_list)
    
    # 初始化结果矩阵（1500x1500仅需~11MB内存，用int32足够）
    result_matrix = np.zeros((num_files, num_files), dtype=np.int32)
    
    # 生成无向文件对（避免重复计算x-y和y-x）
    file_pairs = []
    for i in range(num_files):
        for j in range(i+1, num_files):
            file_pairs.append((file_list[i], file_list[j], npy_dir))
    
    # 并行处理（进程数设为CPU核心数，平衡IO与CPU）
    with ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
        for file_x, file_y, count in executor.map(process_file_pair, file_pairs):
            idx_x = file_list.index(file_x)
            idx_y = file_list.index(file_y)
            result_matrix[idx_x][idx_y] = count
            result_matrix[idx_y][idx_x] = count
    
    # 保存结果矩阵
    np.save("/path/to/final_result_matrix.npy", result_matrix)

if __name__ == "__main__":
    main()

关键库推荐 #

1. numpy：核心工具，提供高效的数组存储、排序、内存映射功能，是整个方案的基础。
2. gzip/lzma/bz2：Python标准库，用于读取不同格式的压缩文件。
3. concurrent.futures：简化多进程并行处理，避开GIL对CPU密集型任务的限制。
4. gc（可选）：手动触发垃圾回收，在多进程场景下避免内存泄漏。

额外优化技巧 #

• 整数类型选择：用np.int32存储整数（如果范围在-2^31~231-1之间），比np.int64省一半内存。
• 二进制原始文件：如果原始压缩文件是二进制格式（而非文本），用np.fromfile读取，速度比逐行读取文本快数倍。
• 任务分批：如果文件对数量过大，可将任务分成批次处理，避免一次性生成所有对占用过多内存。
• 避免np.intersect1d：该函数会生成交集数组，占用额外内存，双指针法更适合大数组的计数需求。

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