本文介绍如何在内存受限条件下，对多个已排序的大文本文件进行流式逐行读取与 k 路归并，支持重复键的值累加（如求和），避免一次性加载文件，真正实现低内存占用的磁盘文件合并。
本文介绍如何在内存受限条件下，对多个已排序的大文本文件进行流式逐行读取与 k 路归并，支持重复键的值累加（如求和），避免一次性加载文件，真正实现低内存占用的磁盘文件合并。
在处理超大规模键值数据（如日志聚合、分布式计算中间结果）时，常见场景是：多个已按 key 排序的纯文本文件（每行格式为 key\tvalue），单个文件远大于可用内存，需合并为一个全局有序且同 key 值合并（如 sum）的结果文件。此时，不能使用 readlines() 或 list(file)——它们会将整个文件载入内存；也不能依赖 for line in file 的隐式迭代器（因需并发控制多文件读取位置）。核心突破口在于：显式调用 file.readline() 实现对每个文件的细粒度、按需行读取。
✅ 正确做法：基于 readline() 的 k 路归并
我们使用最小堆（heapq）维护当前各文件的“前沿”行，每次弹出最小 key 的条目，并从对应文件读取下一行，同时对相同 key 进行归约（如累加 value）。关键点：
- 每个文件保持打开状态，仅用 readline() 获取下一行；
- 使用 (key, value, file_handle, file_id) 元组入堆，确保稳定排序；
- 遇到连续相同 key 时，在堆外暂存并聚合，避免重复入堆。
以下为完整可运行示例（假设所有文件均为 \t 分隔，value 为数字）：
import heapq
import sys
def merge_sorted_files(file_paths, output_path, value_func=sum):
"""
合并多个已排序的键值文件（key\tvalue 格式），相同 key 的 value 被聚合（默认求和）
:param file_paths: 文件路径列表，每个文件按 key 升序排列
:param output_path: 输出文件路径
:param value_func: 聚合函数，接收 value 列表，返回聚合结果（如 sum, max, list）
"""
# 打开所有输入文件
files = [open(path, 'r', encoding='utf-8') for path in file_paths]
heap = []
# 初始化：读取每个文件第一行，推入堆
for i, f in enumerate(files):
line = f.readline().rstrip('\n')
if line:
try:
key, val_str = line.split('\t', 1)
value = float(val_str) if '.' in val_str else int(val_str)
heapq.heappush(heap, (key, value, f, i))
except (ValueError, IndexError) as e:
raise ValueError(f"Invalid line in {file_paths[i]}: {line!r}") from e
# 归并主循环
with open(output_path, 'w', encoding='utf-8') as out:
prev_key = None
accumulated_values = []
while heap:
key, value, f, idx = heapq.heappop(heap)
# 累积相同 key 的 value
if key == prev_key:
accumulated_values.append(value)
else:
# 输出上一组结果（若存在）
if accumulated_values:
result_value = value_func(accumulated_values)
out.write(f"{prev_key}\t{result_value}\n")
# 重置当前 key
prev_key = key
accumulated_values = [value]
# 从同一文件读取下一行，继续入堆
next_line = f.readline().rstrip('\n')
if next_line:
try:
next_key, next_val_str = next_line.split('\t', 1)
next_val = float(next_val_str) if '.' in next_val_str else int(next_val_str)
heapq.heappush(heap, (next_key, next_val, f, idx))
except (ValueError, IndexError) as e:
raise ValueError(f"Invalid line in {file_paths[idx]}: {next_line!r}") from e
# 输出最后一组
if accumulated_values:
result_value = value_func(accumulated_values)
out.write(f"{prev_key}\t{result_value}\n")
# 关闭所有文件
for f in files:
f.close()
# 使用示例
if __name__ == "__main__":
merge_sorted_files(
file_paths=['data_001.txt', 'data_002.txt', 'data_003.txt'],
output_path='merged_output.txt',
value_func=sum
)⚠️ 注意事项与最佳实践
- 编码与分隔符：务必确认文件编码（推荐 UTF-8）及字段分隔符（示例中为 \t，可根据需要替换为 , 或空格）；
- 内存安全：readline() 每次仅读取一行（含换行符），内存占用与最长行长度成正比，而非文件总大小；
- 错误处理：示例中包含基础解析异常捕获，生产环境建议增加日志记录与部分失败恢复机制；
- 扩展性：如需支持自定义 value 类型（如 JSON 对象），可将 value_func 改为接受解析器函数（如 json.loads）；
- 性能优化：对极多文件（>100），可考虑分批归并（两两合并再递归），降低堆操作开销；
- 资源管理：使用 contextlib.ExitStack 可更优雅地管理大量文件句柄（尤其在异常路径下）。
该方案完全规避了 for line in file 的不可控迭代特性，以 readline() 为基石，结合堆排序与流式归约，实现了真正面向大数据场景的、内存友好的磁盘文件合并。
立即学习“Python免费学习笔记（深入）”；
