本文介绍如何在不将大文件全部加载到内存的前提下，利用readline()逐行读取多个已排序的键值文件，并完成k路归并及相同键的值累加。
本文介绍如何在不将大文件全部加载到内存的前提下，利用readline()逐行读取多个已排序的键值文件，并完成k路归并及相同键的值累加。
在处理超大文本文件（如日志、分片导出的键值数据）时，常见的需求是将多个已按key升序排列的文件合并为一个全局有序的结果，同时对重复key的value进行聚合（例如求和）。由于文件体积远超可用内存，不能使用readlines()或list(file)一次性加载，也不能依赖for line in file隐式迭代——因为我们需要对多个文件手控读取节奏，以实现同步比较与归并。
核心突破口在于：file.readline() 是唯一能按需触发单行读取、且不预加载后续内容的标准方法。它返回下一行字符串（含换行符），到达EOF时返回空字符串，完全符合外部归并控制流的需求。
以下是一个完整、健壮的实现示例：
import heapq
from typing import List, Tuple, Optional, TextIO
def merge_sorted_kv_files(file_paths: List[str], output_path: str, delimiter: str = '\t') -> None:
"""
合并多个已按key排序的键值文件，相同key的value相加，输出到新文件。
:param file_paths: 输入文件路径列表（每个文件每行格式：key{delimiter}value）
:param output_path: 输出文件路径
:param delimiter: 键值分隔符，默认为制表符
"""
# 打开所有输入文件
files = [open(path, 'r', encoding='utf-8') for path in file_paths]
try:
# 初始化：为每个文件读取首行，构建最小堆（key, value, file_index）
heap = []
lines = [None] * len(files) # 缓存当前各行内容（避免重复readline）
for i, f in enumerate(files):
line = f.readline().strip()
if line:
parts = line.split(delimiter, 1)
if len(parts) < 2:
raise ValueError(f"Invalid line format in {file_paths[i]}: {line}")
key, val_str = parts[0].strip(), parts[1].strip()
try:
val = float(val_str) # 支持整数/浮点数；按需可改为int或自定义解析
except ValueError:
raise ValueError(f"Non-numeric value in {file_paths[i]}: {val_str}")
heapq.heappush(heap, (key, val, i))
lines[i] = line # 缓存该行原始字符串（用于后续重读？不必要，但记录状态）
else:
# 文件为空，跳过
pass
# 归并主循环
with open(output_path, 'w', encoding='utf-8') as out_f:
while heap:
# 取出当前最小key
current_key, current_sum, idx = heapq.heappop(heap)
# 合并所有相同key的条目
while heap and heap[0][0] == current_key:
_, val, i = heapq.heappop(heap)
current_sum += val
# 从对应文件读下一行
next_line = files[i].readline().strip()
if next_line:
parts = next_line.split(delimiter, 1)
if len(parts) < 2:
raise ValueError(f"Invalid line in {file_paths[i]}: {next_line}")
k, v_str = parts[0].strip(), parts[1].strip()
try:
v = float(v_str)
except ValueError:
raise ValueError(f"Non-numeric value in {file_paths[i]}: {v_str}")
heapq.heappush(heap, (k, v, i))
# 写入合并结果
out_f.write(f"{current_key}{delimiter}{current_sum}\n")
# 为刚消耗掉的文件（idx）补充下一行
next_line = files[idx].readline().strip()
if next_line:
parts = next_line.split(delimiter, 1)
if len(parts) < 2:
raise ValueError(f"Invalid line in {file_paths[idx]}: {next_line}")
k, v_str = parts[0].strip(), parts[1].strip()
try:
v = float(v_str)
except ValueError:
raise ValueError(f"Non-numeric value in {file_paths[idx]}: {v_str}")
heapq.heappush(heap, (k, v, idx))
finally:
# 确保所有文件关闭
for f in files:
f.close()
# 使用示例
if __name__ == "__main__":
merge_sorted_kv_files(
file_paths=["data_001.txt", "data_002.txt", "data_003.txt"],
output_path="merged_output.txt"
)✅ 关键设计说明：
立即学习“Python免费学习笔记（深入）”；
- 使用 heapq 实现k路归并，时间复杂度 O(N log k)，其中 N 为总行数，k 为文件数；
- 每次仅用 readline() 获取一行，内存占用恒定（与文件数量成正比，与文件大小无关）；
- 对相同key自动聚合（此处为累加），支持浮点数值；
- 包含基础错误校验（格式、数值合法性），便于生产环境调试；
- 显式管理文件句柄，确保异常时也能安全关闭（try/finally）。
⚠️ 注意事项：
- 所有输入文件必须严格按key升序排列，否则归并结果不正确；
- 若value需其他聚合逻辑（如取最大值、拼接字符串），修改内部current_sum累加部分即可；
- 如需更高性能，可考虑用mmap或pandas.read_csv(chunksize=...)替代纯文本处理，但会增加依赖与复杂度；
- Python 3.12+ 中可结合 contextlib.aclosing 或 asyncio 实现异步I/O，进一步提升吞吐量（适用于SSD/NVMe场景）。
该方案平衡了简洁性、可维护性与工程鲁棒性，是处理TB级分片键值数据合并任务的推荐实践。
