代码收藏家技术教程 2025-05-20

【Python实战】解决TB级别大文件自然数排序难题：解决方案详解及完整代码示例

📌 一、问题背景

输入数据： 一个大小为 100TB 的文本文件，每行一个自然数。

可用资源： 内存限制为 16GB。

目标： 对这 100TB 的数据进行升序排序并输出到文件。

由于数据量远大于内存容量，无法一次性加载全部数据进行排序，因此需要采用分治 + 外排序的方法来解决。

🧩 二、解决方案概述

整个流程分为四个核心步骤：

查找最大值： 遍历原始文件找到最大数值，用于后续合理划分区间。
分片（Partitioning）： 按照数值范围将大文件拆分成多个小文件。
排序（Internal Sorting）： 将每个小文件加载进内存排序后写入磁盘。
多路归并（K-way Merge）： 使用最小堆算法合并所有已排序的小文件，生成最终有序结果。

🛠️ 三、具体实现步骤详解

✅ 步骤 1：查找最大值

def find_max_number(file_path):
    max_num = 0
    with open(file_path, 'r') as f:
        for line in f:
            num = int(line.strip())
            if num > max_num:
                max_num = num
    return max_num

🔍 说明：

这一步是为了动态设置数值区间的大小，确保分区均匀，避免出现某些分区过大或为空的情况。

✅ 步骤 2：按数值区间分片

def partition_file(max_num):
    range_size = max_num // NUM_PARTITIONS + 1
    partition_files = [
        open(os.path.join(PARTITION_DIR, f'partition_{i}.txt'), 'w')
        for i in range(NUM_PARTITIONS)
    ]

    with open(INPUT_FILE, 'r') as f:
        for line in f:
            num = int(line.strip())
            idx = num // range_size
            if idx >= NUM_PARTITIONS:
                idx = NUM_PARTITIONS - 1
            partition_files[idx].write(line)

    for fp in partition_files:
        fp.close()

🔍 说明：

使用 num // range_size 计算该数字属于哪个分区，从而将原始数据分散到多个文件中，便于后续排序。

✅ 步骤 3：内部排序每个小文件

def sort_partitions():
    for i in range(NUM_PARTITIONS):
        input_path = os.path.join(PARTITION_DIR, f'partition_{i}.txt')
        output_path = os.path.join(PARTITION_DIR, f'sorted_partition_{i}.txt')

        with open(input_path, 'r') as f:
            nums = list(map(int, f.readlines()))
        nums.sort()

        with open(output_path, 'w') as f:
            for num in nums:
                f.write(str(num) + '\n')

🔍 说明：

每个小文件的数据量控制在 1GB 左右，可以轻松加载进内存进行排序。

✅ 步骤 4：多路归并排序（K-way Merge）

def merge_sorted_partitions():
    heap = []
    pointers = []

    for i in range(NUM_PARTITIONS):
        path = os.path.join(PARTITION_DIR, f'sorted_partition_{i}.txt')
        fp = open(path, 'r')
        line = fp.readline()
        if line:
            heapq.heappush(heap, (int(line.strip()), i))
        pointers.append(fp)

    with open(OUTPUT_FILE, 'w') as out_file:
        while heap:
            val, idx = heapq.heappop(heap)
            out_file.write(str(val) + '\n')

            next_line = pointers[idx].readline()
            if next_line:
                heapq.heappush(heap, (int(next_line.strip()), idx))

    for fp in pointers:
        fp.close()

🔍 说明：

使用 最小堆 来高效地从多个有序文件中取出当前最小值，逐步构建最终有序序列。这是大数据归并的经典做法。

📈 四、时间复杂度与性能分析

步骤	时间复杂度	说明
查找最大值	O(N)	遍历一次文件
分片	O(N)	每个数字只处理一次
排序各分片	O(n log n) × K	每个分片排序
多路归并	O(N log K)	使用堆结构高效归并

整体效率高，适合处理超大规模数据。

🧪 五、测试脚本：生成 1GB 测试数据

为了验证程序是否正常工作，我们可以先生成一个 1GB 的测试文件：

import random

TARGET_SIZE_BYTES = 1 * 1024 * 1024 * 1024  # 1GB
CHUNK_SIZE = 100000

with open('input.txt', 'w') as f:
    total = 0
    while total < TARGET_SIZE_BYTES:
        data = '\n'.join(str(random.randint(0, 10**9)) for _ in range(CHUNK_SIZE)) + '\n'
        f.write(data)
        total += len(data.encode())
        print(f"\r已写入: {total // (1024 * 1024)} MB", end='')